aminoacid
1#!/usr/bin/env python 2from __future__ import annotations 3 4import asyncio 5from functools import wraps 6from importlib.util import find_spec 7from logging import Logger, getLogger 8from secrets import token_urlsafe 9from shlex import split 10from time import asctime 11from typing import Any, Callable, Coroutine, Dict, List, Optional, TypeVar 12 13from aiohttp import ClientResponse, ClientSession 14 15from ._socket import SocketClient 16from .abc import _ORJSON, Context, Message, Session, User, json 17from .client import ApiClient 18from .exceptions import AminoBaseException, CommandExists, CommandNotFound 19from .util import HelpCommand, UserCommand, __version__, get_headers 20 21__author__ = "okok7711" 22 23_COLOR = find_spec("coloredlogs") 24 25if _COLOR: 26 import coloredlogs 27 28 coloredlogs.install() 29 30T = TypeVar("T") 31 32json_serialize = ( 33 lambda obj, *args, **kwargs: json.dumps(obj).decode() 34 if _ORJSON 35 else json.dumps(obj) 36) 37"""Function to serialize data with ORJson if its installed, uses built in json module if not 38 39Parameters 40---------- 41obj : dict 42 object to serialize 43 44Returns 45------- 46str 47 serialized json data 48""" 49 50 51class Bot(ApiClient): 52 r"""Bot client class, this is the interface for doing anything 53 54 Parameters 55 ---------- 56 key : bytes 57 The key to use for signing requests. This has to be supplied in kwargs otherwise the Bot will not work 58 device : str 59 The deviceId to use. This is also needed for the bot to run 60 prefix : str, optional 61 The prefix the bot should listen to. By default "/" 62 help_command : UserCommand, optional 63 The help_command to use, if not supplied a standard Help Command will be used. 64 v : bytes, optional 65 The version of the signing algorithm. This is currently \x42 66 sessionId : str, optional 67 If given, the bot will not auth via login but will use the given session instead 68 """ 69 loop: asyncio.AbstractEventLoop 70 profile: User 71 72 def __init__( 73 self, prefix: str = "/", *, help_command: Optional[UserCommand] = None, **kwargs 74 ) -> None: 75 self.prefix = prefix 76 self.__command_map__: Dict[str, UserCommand] = { 77 "help": help_command or HelpCommand() 78 } 79 self.events: Dict[str, Callable[..., Coroutine[Any, Any, T]]] = {} 80 self.logger = getLogger(__name__) 81 self._http = HttpClient(logger=self.logger, **kwargs) 82 83 super().__init__() 84 85 def unregister_command(self, name: str): 86 if not self.__command_map__.pop(name, ""): 87 raise CommandNotFound(name) 88 89 def command( 90 self, 91 name="", 92 *, 93 check: Optional[Callable[[Context], bool]] = lambda _: True, 94 check_any: Optional[List[Callable[[Context], bool]]] = [], 95 cooldown: Optional[int] = 0, 96 ): 97 """Wrapper to register commands to the bot 98 99 Parameters 100 ---------- 101 name : str, optional 102 Name that the command should listen on, by default the name of the function 103 check : Optional[Callable[[Context], bool]], optional 104 function to check whether the command may be executed, by default always returns True 105 check_any : Optional[List[Callable[[Context], bool]]], optional 106 list of checks of which any need to pass, by default [] 107 cooldown : Optional[int], optional 108 cooldown before someone can use the command again, by default 0 109 110 Returns 111 ------- 112 UserCommand 113 returns the command to register error handlers etc. 114 """ 115 116 def wrap(f: Callable[..., Coroutine[Any, Any, T]]): 117 @wraps(f) 118 def func() -> UserCommand: 119 if (name or f.__name__) in self.__command_map__: 120 return self.logger.exception(CommandExists(name)) 121 self.__command_map__[name or f.__name__] = ( 122 cmd := UserCommand( 123 func=f, 124 command_name=name, 125 check=check, 126 check_any=check_any, 127 cooldown=cooldown, 128 ) 129 ) 130 return cmd 131 132 return func() 133 134 return wrap 135 136 def event(self, name=""): 137 """Wrapper to add events the bot should listen to 138 All possible events are defined in `aminoacid.util.events` 139 140 Parameters 141 ---------- 142 name : str, optional 143 Name that the event should listen on, by default the name of the function 144 """ 145 146 def wrap(f: Callable): 147 @wraps(f) 148 def func(): 149 if (name or f.__name__) in self.events: 150 return 151 self.events[name or f.__name__] = f 152 153 return func() 154 155 return wrap 156 157 def run( 158 self, 159 email: Optional[str] = "", 160 password: Optional[str] = "", 161 *, 162 session: str = "", 163 ): 164 """Run the `main_loop()` of the bot and initiate authentication 165 166 Parameters 167 ---------- 168 email : str, optional 169 email of the account to use 170 password : str, optional 171 password of the account to use 172 session : str, optional 173 session to use instead of email password combination 174 """ 175 if not any((email, password, session)): 176 raise Exception("No Auth") 177 self.loop = asyncio.get_event_loop() 178 self.loop.run_until_complete( 179 self.main_loop(email=email, password=password, sessionId=session) 180 ) 181 182 async def main_loop( 183 self, 184 email: Optional[str], 185 password: Optional[str], 186 *, 187 sessionId: Optional[str] = "", 188 ): 189 """Main loop of the Bot, this authenticates and sets the session 190 191 Parameters 192 ---------- 193 email : Optional[str] 194 Email of the account to login 195 password : Optional[str] 196 Password of the account to login 197 sessionId : Optional[str], optional 198 Session of the Account, by default "" 199 """ 200 if not sessionId: 201 await self.login(email=email, password=password) 202 else: 203 self._http.session = Session(sessionId) 204 self.profile = await self.fetch_user(self._http.session.uid) 205 self.socket = sock = SocketClient(self) 206 try: 207 await self.set_device() 208 await sock.run_loop() 209 finally: 210 await self.cleanup_loop() 211 212 async def handle_command(self, message: Message): 213 """Handles a command for the supplied message 214 215 Parameters 216 ---------- 217 message : Message 218 The message to handle 219 """ 220 if not message.startswith(self.prefix): 221 return 222 args = split(message.content[len(self.prefix) :]) 223 if args[0] in self.__command_map__: 224 coro = (cmd := self.__command_map__[args.pop(0)])( 225 (ctx := Context(client=self, message=message)), *args 226 ) 227 if coro: 228 try: 229 await coro 230 except AminoBaseException as exc: 231 if cmd.handler: 232 await cmd.handler(exc, ctx) 233 else: 234 self.logger.exception(exc) 235 else: 236 self.logger.exception(CommandNotFound(message)) 237 238 async def cleanup_loop(self): 239 """Cleans up, closes sessions, etc.""" 240 # TODO: maybe make better cleanup 241 await self._http.close() 242 243 244class HttpClient(ClientSession): 245 session: Session 246 247 def __init__(self, logger: Logger, *args, **kwargs) -> None: 248 self.base: str = kwargs.pop("base_uri", "https://service.narvii.com/api/v1") 249 self.key: bytes = kwargs.pop("key") 250 self.device: str = kwargs.pop("device") 251 self.v: bytes = kwargs.pop("v", b"\x42") 252 self.token: str = token_urlsafe(152) 253 """pushToken for notification sending, randomly generated""" 254 255 self.logger = logger 256 257 self.session = None 258 259 super().__init__(*args, **kwargs, json_serialize=json_serialize) 260 261 async def request(self, method: str, url: str, *args, **kwargs) -> ClientResponse: 262 """Execute a request 263 264 Parameters 265 ---------- 266 method : str 267 Method of the request 268 url : str 269 url to be appended to the base 270 271 Returns 272 ------- 273 ClientResponse 274 Response the server returned 275 """ 276 277 headers = kwargs.pop("headers", {}) or get_headers( 278 data=json_serialize(kwargs.get("json", {})).encode(), 279 device=self.device, 280 key=self.key, 281 v=self.v, 282 ) 283 if self.session: 284 headers["NDCAUTH"] = self.session.sid 285 response = await super().request( 286 method, 287 url=(self.base + url if not "wss" in url else url), 288 headers=headers, 289 *args, 290 **kwargs, 291 ) 292 await self.log_request(response, self.logger) 293 return response 294 295 @staticmethod 296 async def log_request(response: ClientResponse, logger: Logger) -> None: 297 """Logs a request and its response with info level 298 299 Parameters 300 ---------- 301 response : ClientResponse 302 the response object that the request returned 303 """ 304 logger.info( 305 f"{response.request_info.method} [{asctime()}] -> {response.url}: {response.status} [{response.content_type}] " 306 f"Received Headers: {response.headers}, Sent Headers: {response.request_info.headers}, Received Content: {await response.text()}" 307 )
def
json_serialize(obj, *args, **kwargs):
34 lambda obj, *args, **kwargs: json.dumps(obj).decode()
Function to serialize data with ORJson if its installed, uses built in json module if not
Parameters
- obj (dict): object to serialize
Returns
- str: serialized json data
52class Bot(ApiClient): 53 r"""Bot client class, this is the interface for doing anything 54 55 Parameters 56 ---------- 57 key : bytes 58 The key to use for signing requests. This has to be supplied in kwargs otherwise the Bot will not work 59 device : str 60 The deviceId to use. This is also needed for the bot to run 61 prefix : str, optional 62 The prefix the bot should listen to. By default "/" 63 help_command : UserCommand, optional 64 The help_command to use, if not supplied a standard Help Command will be used. 65 v : bytes, optional 66 The version of the signing algorithm. This is currently \x42 67 sessionId : str, optional 68 If given, the bot will not auth via login but will use the given session instead 69 """ 70 loop: asyncio.AbstractEventLoop 71 profile: User 72 73 def __init__( 74 self, prefix: str = "/", *, help_command: Optional[UserCommand] = None, **kwargs 75 ) -> None: 76 self.prefix = prefix 77 self.__command_map__: Dict[str, UserCommand] = { 78 "help": help_command or HelpCommand() 79 } 80 self.events: Dict[str, Callable[..., Coroutine[Any, Any, T]]] = {} 81 self.logger = getLogger(__name__) 82 self._http = HttpClient(logger=self.logger, **kwargs) 83 84 super().__init__() 85 86 def unregister_command(self, name: str): 87 if not self.__command_map__.pop(name, ""): 88 raise CommandNotFound(name) 89 90 def command( 91 self, 92 name="", 93 *, 94 check: Optional[Callable[[Context], bool]] = lambda _: True, 95 check_any: Optional[List[Callable[[Context], bool]]] = [], 96 cooldown: Optional[int] = 0, 97 ): 98 """Wrapper to register commands to the bot 99 100 Parameters 101 ---------- 102 name : str, optional 103 Name that the command should listen on, by default the name of the function 104 check : Optional[Callable[[Context], bool]], optional 105 function to check whether the command may be executed, by default always returns True 106 check_any : Optional[List[Callable[[Context], bool]]], optional 107 list of checks of which any need to pass, by default [] 108 cooldown : Optional[int], optional 109 cooldown before someone can use the command again, by default 0 110 111 Returns 112 ------- 113 UserCommand 114 returns the command to register error handlers etc. 115 """ 116 117 def wrap(f: Callable[..., Coroutine[Any, Any, T]]): 118 @wraps(f) 119 def func() -> UserCommand: 120 if (name or f.__name__) in self.__command_map__: 121 return self.logger.exception(CommandExists(name)) 122 self.__command_map__[name or f.__name__] = ( 123 cmd := UserCommand( 124 func=f, 125 command_name=name, 126 check=check, 127 check_any=check_any, 128 cooldown=cooldown, 129 ) 130 ) 131 return cmd 132 133 return func() 134 135 return wrap 136 137 def event(self, name=""): 138 """Wrapper to add events the bot should listen to 139 All possible events are defined in `aminoacid.util.events` 140 141 Parameters 142 ---------- 143 name : str, optional 144 Name that the event should listen on, by default the name of the function 145 """ 146 147 def wrap(f: Callable): 148 @wraps(f) 149 def func(): 150 if (name or f.__name__) in self.events: 151 return 152 self.events[name or f.__name__] = f 153 154 return func() 155 156 return wrap 157 158 def run( 159 self, 160 email: Optional[str] = "", 161 password: Optional[str] = "", 162 *, 163 session: str = "", 164 ): 165 """Run the `main_loop()` of the bot and initiate authentication 166 167 Parameters 168 ---------- 169 email : str, optional 170 email of the account to use 171 password : str, optional 172 password of the account to use 173 session : str, optional 174 session to use instead of email password combination 175 """ 176 if not any((email, password, session)): 177 raise Exception("No Auth") 178 self.loop = asyncio.get_event_loop() 179 self.loop.run_until_complete( 180 self.main_loop(email=email, password=password, sessionId=session) 181 ) 182 183 async def main_loop( 184 self, 185 email: Optional[str], 186 password: Optional[str], 187 *, 188 sessionId: Optional[str] = "", 189 ): 190 """Main loop of the Bot, this authenticates and sets the session 191 192 Parameters 193 ---------- 194 email : Optional[str] 195 Email of the account to login 196 password : Optional[str] 197 Password of the account to login 198 sessionId : Optional[str], optional 199 Session of the Account, by default "" 200 """ 201 if not sessionId: 202 await self.login(email=email, password=password) 203 else: 204 self._http.session = Session(sessionId) 205 self.profile = await self.fetch_user(self._http.session.uid) 206 self.socket = sock = SocketClient(self) 207 try: 208 await self.set_device() 209 await sock.run_loop() 210 finally: 211 await self.cleanup_loop() 212 213 async def handle_command(self, message: Message): 214 """Handles a command for the supplied message 215 216 Parameters 217 ---------- 218 message : Message 219 The message to handle 220 """ 221 if not message.startswith(self.prefix): 222 return 223 args = split(message.content[len(self.prefix) :]) 224 if args[0] in self.__command_map__: 225 coro = (cmd := self.__command_map__[args.pop(0)])( 226 (ctx := Context(client=self, message=message)), *args 227 ) 228 if coro: 229 try: 230 await coro 231 except AminoBaseException as exc: 232 if cmd.handler: 233 await cmd.handler(exc, ctx) 234 else: 235 self.logger.exception(exc) 236 else: 237 self.logger.exception(CommandNotFound(message)) 238 239 async def cleanup_loop(self): 240 """Cleans up, closes sessions, etc.""" 241 # TODO: maybe make better cleanup 242 await self._http.close()
Bot client class, this is the interface for doing anything
Parameters
- key (bytes): The key to use for signing requests. This has to be supplied in kwargs otherwise the Bot will not work
- device (str): The deviceId to use. This is also needed for the bot to run
- prefix (str, optional): The prefix the bot should listen to. By default "/"
- help_command (UserCommand, optional): The help_command to use, if not supplied a standard Help Command will be used.
- v (bytes, optional): The version of the signing algorithm. This is currently \x42
- sessionId (str, optional): If given, the bot will not auth via login but will use the given session instead
Bot( prefix: str = '/', *, help_command: Optional[aminoacid.util.commands.UserCommand] = None, **kwargs)
73 def __init__( 74 self, prefix: str = "/", *, help_command: Optional[UserCommand] = None, **kwargs 75 ) -> None: 76 self.prefix = prefix 77 self.__command_map__: Dict[str, UserCommand] = { 78 "help": help_command or HelpCommand() 79 } 80 self.events: Dict[str, Callable[..., Coroutine[Any, Any, T]]] = {} 81 self.logger = getLogger(__name__) 82 self._http = HttpClient(logger=self.logger, **kwargs) 83 84 super().__init__()
def
command( self, name='', *, check: Optional[Callable[[aminoacid.abc.Context], bool]] = <function Bot.<lambda>>, check_any: Optional[List[Callable[[aminoacid.abc.Context], bool]]] = [], cooldown: Optional[int] = 0):
90 def command( 91 self, 92 name="", 93 *, 94 check: Optional[Callable[[Context], bool]] = lambda _: True, 95 check_any: Optional[List[Callable[[Context], bool]]] = [], 96 cooldown: Optional[int] = 0, 97 ): 98 """Wrapper to register commands to the bot 99 100 Parameters 101 ---------- 102 name : str, optional 103 Name that the command should listen on, by default the name of the function 104 check : Optional[Callable[[Context], bool]], optional 105 function to check whether the command may be executed, by default always returns True 106 check_any : Optional[List[Callable[[Context], bool]]], optional 107 list of checks of which any need to pass, by default [] 108 cooldown : Optional[int], optional 109 cooldown before someone can use the command again, by default 0 110 111 Returns 112 ------- 113 UserCommand 114 returns the command to register error handlers etc. 115 """ 116 117 def wrap(f: Callable[..., Coroutine[Any, Any, T]]): 118 @wraps(f) 119 def func() -> UserCommand: 120 if (name or f.__name__) in self.__command_map__: 121 return self.logger.exception(CommandExists(name)) 122 self.__command_map__[name or f.__name__] = ( 123 cmd := UserCommand( 124 func=f, 125 command_name=name, 126 check=check, 127 check_any=check_any, 128 cooldown=cooldown, 129 ) 130 ) 131 return cmd 132 133 return func() 134 135 return wrap
Wrapper to register commands to the bot
Parameters
- name (str, optional): Name that the command should listen on, by default the name of the function
- check (Optional[Callable[[Context], bool]], optional): function to check whether the command may be executed, by default always returns True
- check_any (Optional[List[Callable[[Context], bool]]], optional): list of checks of which any need to pass, by default []
- cooldown (Optional[int], optional): cooldown before someone can use the command again, by default 0
Returns
- UserCommand: returns the command to register error handlers etc.
def
event(self, name=''):
137 def event(self, name=""): 138 """Wrapper to add events the bot should listen to 139 All possible events are defined in `aminoacid.util.events` 140 141 Parameters 142 ---------- 143 name : str, optional 144 Name that the event should listen on, by default the name of the function 145 """ 146 147 def wrap(f: Callable): 148 @wraps(f) 149 def func(): 150 if (name or f.__name__) in self.events: 151 return 152 self.events[name or f.__name__] = f 153 154 return func() 155 156 return wrap
Wrapper to add events the bot should listen to
All possible events are defined in aminoacid.util.events
Parameters
- name (str, optional): Name that the event should listen on, by default the name of the function
def
run( self, email: Optional[str] = '', password: Optional[str] = '', *, session: str = ''):
158 def run( 159 self, 160 email: Optional[str] = "", 161 password: Optional[str] = "", 162 *, 163 session: str = "", 164 ): 165 """Run the `main_loop()` of the bot and initiate authentication 166 167 Parameters 168 ---------- 169 email : str, optional 170 email of the account to use 171 password : str, optional 172 password of the account to use 173 session : str, optional 174 session to use instead of email password combination 175 """ 176 if not any((email, password, session)): 177 raise Exception("No Auth") 178 self.loop = asyncio.get_event_loop() 179 self.loop.run_until_complete( 180 self.main_loop(email=email, password=password, sessionId=session) 181 )
Run the main_loop() of the bot and initiate authentication
Parameters
- email (str, optional): email of the account to use
- password (str, optional): password of the account to use
- session (str, optional): session to use instead of email password combination
async def
main_loop( self, email: Optional[str], password: Optional[str], *, sessionId: Optional[str] = ''):
183 async def main_loop( 184 self, 185 email: Optional[str], 186 password: Optional[str], 187 *, 188 sessionId: Optional[str] = "", 189 ): 190 """Main loop of the Bot, this authenticates and sets the session 191 192 Parameters 193 ---------- 194 email : Optional[str] 195 Email of the account to login 196 password : Optional[str] 197 Password of the account to login 198 sessionId : Optional[str], optional 199 Session of the Account, by default "" 200 """ 201 if not sessionId: 202 await self.login(email=email, password=password) 203 else: 204 self._http.session = Session(sessionId) 205 self.profile = await self.fetch_user(self._http.session.uid) 206 self.socket = sock = SocketClient(self) 207 try: 208 await self.set_device() 209 await sock.run_loop() 210 finally: 211 await self.cleanup_loop()
Main loop of the Bot, this authenticates and sets the session
Parameters
- email (Optional[str]): Email of the account to login
- password (Optional[str]): Password of the account to login
- sessionId (Optional[str], optional): Session of the Account, by default ""
213 async def handle_command(self, message: Message): 214 """Handles a command for the supplied message 215 216 Parameters 217 ---------- 218 message : Message 219 The message to handle 220 """ 221 if not message.startswith(self.prefix): 222 return 223 args = split(message.content[len(self.prefix) :]) 224 if args[0] in self.__command_map__: 225 coro = (cmd := self.__command_map__[args.pop(0)])( 226 (ctx := Context(client=self, message=message)), *args 227 ) 228 if coro: 229 try: 230 await coro 231 except AminoBaseException as exc: 232 if cmd.handler: 233 await cmd.handler(exc, ctx) 234 else: 235 self.logger.exception(exc) 236 else: 237 self.logger.exception(CommandNotFound(message))
Handles a command for the supplied message
Parameters
- message (Message): The message to handle
class
HttpClient(aiohttp.client.ClientSession):
245class HttpClient(ClientSession): 246 session: Session 247 248 def __init__(self, logger: Logger, *args, **kwargs) -> None: 249 self.base: str = kwargs.pop("base_uri", "https://service.narvii.com/api/v1") 250 self.key: bytes = kwargs.pop("key") 251 self.device: str = kwargs.pop("device") 252 self.v: bytes = kwargs.pop("v", b"\x42") 253 self.token: str = token_urlsafe(152) 254 """pushToken for notification sending, randomly generated""" 255 256 self.logger = logger 257 258 self.session = None 259 260 super().__init__(*args, **kwargs, json_serialize=json_serialize) 261 262 async def request(self, method: str, url: str, *args, **kwargs) -> ClientResponse: 263 """Execute a request 264 265 Parameters 266 ---------- 267 method : str 268 Method of the request 269 url : str 270 url to be appended to the base 271 272 Returns 273 ------- 274 ClientResponse 275 Response the server returned 276 """ 277 278 headers = kwargs.pop("headers", {}) or get_headers( 279 data=json_serialize(kwargs.get("json", {})).encode(), 280 device=self.device, 281 key=self.key, 282 v=self.v, 283 ) 284 if self.session: 285 headers["NDCAUTH"] = self.session.sid 286 response = await super().request( 287 method, 288 url=(self.base + url if not "wss" in url else url), 289 headers=headers, 290 *args, 291 **kwargs, 292 ) 293 await self.log_request(response, self.logger) 294 return response 295 296 @staticmethod 297 async def log_request(response: ClientResponse, logger: Logger) -> None: 298 """Logs a request and its response with info level 299 300 Parameters 301 ---------- 302 response : ClientResponse 303 the response object that the request returned 304 """ 305 logger.info( 306 f"{response.request_info.method} [{asctime()}] -> {response.url}: {response.status} [{response.content_type}] " 307 f"Received Headers: {response.headers}, Sent Headers: {response.request_info.headers}, Received Content: {await response.text()}" 308 )
First-class interface for making HTTP requests.
HttpClient(logger: logging.Logger, *args, **kwargs)
248 def __init__(self, logger: Logger, *args, **kwargs) -> None: 249 self.base: str = kwargs.pop("base_uri", "https://service.narvii.com/api/v1") 250 self.key: bytes = kwargs.pop("key") 251 self.device: str = kwargs.pop("device") 252 self.v: bytes = kwargs.pop("v", b"\x42") 253 self.token: str = token_urlsafe(152) 254 """pushToken for notification sending, randomly generated""" 255 256 self.logger = logger 257 258 self.session = None 259 260 super().__init__(*args, **kwargs, json_serialize=json_serialize)
async def
request( self, method: str, url: str, *args, **kwargs) -> aiohttp.client_reqrep.ClientResponse:
262 async def request(self, method: str, url: str, *args, **kwargs) -> ClientResponse: 263 """Execute a request 264 265 Parameters 266 ---------- 267 method : str 268 Method of the request 269 url : str 270 url to be appended to the base 271 272 Returns 273 ------- 274 ClientResponse 275 Response the server returned 276 """ 277 278 headers = kwargs.pop("headers", {}) or get_headers( 279 data=json_serialize(kwargs.get("json", {})).encode(), 280 device=self.device, 281 key=self.key, 282 v=self.v, 283 ) 284 if self.session: 285 headers["NDCAUTH"] = self.session.sid 286 response = await super().request( 287 method, 288 url=(self.base + url if not "wss" in url else url), 289 headers=headers, 290 *args, 291 **kwargs, 292 ) 293 await self.log_request(response, self.logger) 294 return response
Perform HTTP request.
@staticmethod
async def
log_request( response: aiohttp.client_reqrep.ClientResponse, logger: logging.Logger) -> None:
296 @staticmethod 297 async def log_request(response: ClientResponse, logger: Logger) -> None: 298 """Logs a request and its response with info level 299 300 Parameters 301 ---------- 302 response : ClientResponse 303 the response object that the request returned 304 """ 305 logger.info( 306 f"{response.request_info.method} [{asctime()}] -> {response.url}: {response.status} [{response.content_type}] " 307 f"Received Headers: {response.headers}, Sent Headers: {response.request_info.headers}, Received Content: {await response.text()}" 308 )
Logs a request and its response with info level
Parameters
- response (ClientResponse): the response object that the request returned
Inherited Members
- aiohttp.client.ClientSession
- ATTRS
- ws_connect
- get
- options
- head
- post
- put
- patch
- delete
- close
- closed
- connector
- version
- requote_redirect_url
- loop
- timeout
- headers
- skip_auto_headers
- auth
- json_serialize
- connector_owner
- raise_for_status
- auto_decompress
- trust_env
- trace_configs
- detach