Reference¶
TODO: get better this page
Connector and Client Client¶
- class aiosonic.connectors.TCPConnector(pool_size: int = 25, timeouts: Optional[aiosonic.timeout.Timeouts] = None, connection_cls=None, pool_cls=None, resolver=None, ttl_dns_cache=10000, use_dns_cache=True)¶
TCPConnector.
Holds the main logic for making connections to destination hosts.
- Params:
pool_size: size for pool of connections
timeouts: global timeouts to use for connections with this connector. default:
aiosonic.timeout.Timeouts
instance with default args.connection_cls: connection class to be used. default:
aiosonic.connection.Connection
pool_cls: pool class to be used. default:
aiosonic.pools.SmartPool
resolver: resolver to be used. default:
aiosonic.resolver.DefaultResolver
ttl_dns_cache: ttl in milliseconds for dns cache. default: 10000 10 seconds
use_dns_cache: Flag to indicate usage of dns cache. default: True
- class aiosonic.HTTPClient(connector: Optional[aiosonic.connectors.TCPConnector] = None, handle_cookies=False, verify_ssl=True, proxy: Optional[aiosonic.proxy.Proxy] = None)¶
aiosonic.HTTPClient class.
This class holds the client creation that will be used for requests.
- Params:
connector: TCPConnector to be used if provided
- handle_cookies: Flag to indicate if keep response cookies in
client and send them in next requests.
verify_ssl: Flag to indicate if verify ssl certificates.
- async aiosonic.HTTPClient.request(self, url: str, method: str = 'GET', headers: Optional[Union[Dict[str, str], List[Tuple[str, str]], aiosonic.HttpHeaders]] = None, params: Optional[Union[Dict[str, str], Sequence[Tuple[str, str]]]] = None, data: Optional[Union[str, bytes, dict, tuple, AsyncIterator[bytes], Iterator[bytes]]] = None, multipart: bool = False, verify: bool = True, ssl: Optional[ssl.SSLContext] = None, timeouts: Optional[aiosonic.timeout.Timeouts] = None, follow: bool = False, http2: bool = False) aiosonic.HttpResponse ¶
Do http request.
- Params:
url: url of request
method: Http method of request
headers: headers to add in request
params: query params to add in request if not manually added
data: Data to be sent, this param is ignored for get requests.
multipart: Tell aiosonic if request is multipart
verify: parameter to indicate whether to verify ssl
ssl: this parameter allows to specify a custom ssl context
timeouts: parameter to indicate timeouts for request
follow: parameter to indicate whether to follow redirects
http2: flag to indicate whether to use http2 (experimental)
- async aiosonic.HTTPClient.get(self, url: str, headers: Optional[Union[Dict[str, str], List[Tuple[str, str]], aiosonic.HttpHeaders]] = None, params: Optional[Union[Dict[str, str], Sequence[Tuple[str, str]]]] = None, verify: bool = True, ssl: Optional[ssl.SSLContext] = None, timeouts: Optional[aiosonic.timeout.Timeouts] = None, follow: bool = False, http2: bool = False) aiosonic.HttpResponse ¶
Do get http request.
- async aiosonic.HTTPClient.post(self, url: str, data: Optional[Union[str, bytes, dict, tuple, AsyncIterator[bytes], Iterator[bytes]]] = None, headers: Optional[Union[Dict[str, str], List[Tuple[str, str]], aiosonic.HttpHeaders]] = None, json: Optional[Union[dict, list]] = None, params: Optional[Union[Dict[str, str], Sequence[Tuple[str, str]]]] = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: Optional[ssl.SSLContext] = None, timeouts: Optional[aiosonic.timeout.Timeouts] = None, follow: bool = False, http2: bool = False) aiosonic.HttpResponse ¶
Do post http request.
- async aiosonic.HTTPClient.put(self, url: str, data: Optional[Union[str, bytes, dict, tuple, AsyncIterator[bytes], Iterator[bytes]]] = None, headers: Optional[Union[Dict[str, str], List[Tuple[str, str]], aiosonic.HttpHeaders]] = None, json: Optional[Union[dict, list]] = None, params: Optional[Union[Dict[str, str], Sequence[Tuple[str, str]]]] = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: Optional[ssl.SSLContext] = None, timeouts: Optional[aiosonic.timeout.Timeouts] = None, follow: bool = False, http2: bool = False) aiosonic.HttpResponse ¶
Do put http request.
- async aiosonic.HTTPClient.patch(self, url: str, data: Optional[Union[str, bytes, dict, tuple, AsyncIterator[bytes], Iterator[bytes]]] = None, headers: Optional[Union[Dict[str, str], List[Tuple[str, str]], aiosonic.HttpHeaders]] = None, json: Optional[Union[dict, list]] = None, params: Optional[Union[Dict[str, str], Sequence[Tuple[str, str]]]] = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: Optional[ssl.SSLContext] = None, timeouts: Optional[aiosonic.timeout.Timeouts] = None, follow: bool = False, http2: bool = False) aiosonic.HttpResponse ¶
Do patch http request.
- async aiosonic.HTTPClient.delete(self, url: str, data: Union[str, bytes, dict, tuple, AsyncIterator[bytes], Iterator[bytes]] = b'', headers: Optional[Union[Dict[str, str], List[Tuple[str, str]], aiosonic.HttpHeaders]] = None, json: Optional[Union[dict, list]] = None, params: Optional[Union[Dict[str, str], Sequence[Tuple[str, str]]]] = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: Optional[ssl.SSLContext] = None, timeouts: Optional[aiosonic.timeout.Timeouts] = None, follow: bool = False, http2: bool = False) aiosonic.HttpResponse ¶
Do delete http request.
- async aiosonic.HTTPClient.wait_requests(self, timeout: int = 30)¶
Wait until all pending requests are done.
If timeout, returns false.
This is useful when doing safe shutdown of a process.
Classes¶
- class aiosonic.HttpHeaders(data=None, **kwargs)¶
Http headers dict.
- class aiosonic.HttpResponse¶
Custom HttpResponse class for handling responses.
- Properties:
status_code (int): response status code
headers (
aiosonic.HttpHeaders
): headers in case insensitive dictcookies (
http.cookies.SimpleCookie
): instance of SimpleCookies if cookies present in respone.raw_headers (List[Tuple[bytes, bytes]]): headers as raw format
- async content() bytes ¶
Read response body.
- async json(json_decoder=<function loads>) dict ¶
Read response body.
- read_chunks() AsyncIterator[bytes] ¶
Read chunks from chunked response.
- property status_code: int¶
Get status code.
- async text() str ¶
Read response body.
Tiemout Class¶
- class aiosonic.timeout.Timeouts(sock_connect: Optional[float] = 5, sock_read: Optional[float] = 30, pool_acquire: Optional[float] = None, request_timeout: Optional[float] = 60)¶
Timeouts class wrapper.
- Arguments:
sock_connect(float): time for establish connection to server
sock_read(float): time until get first read
pool_acquire(float): time until get connection from connection’s pool
request_timeout(float): time until complete request.
Pool Classes¶
- class aiosonic.pools.SmartPool(connector, pool_size, connection_cls)¶
Pool which utilizes alive connections.
- async acquire(urlparsed: Optional[urllib.parse.ParseResult] = None)¶
Acquire connection.
- async cleanup() None ¶
Get all conn and close them, this method let this pool unusable.
- is_all_free()¶
Indicates if all pool is free.
- release(conn) None ¶
Release connection.
- class aiosonic.pools.CyclicQueuePool(connector, pool_size, connection_cls)¶
Cyclic queue pool of connections.
- async acquire(_urlparsed: Optional[urllib.parse.ParseResult] = None)¶
Acquire connection.
- async cleanup()¶
Get all conn and close them, this method let this pool unusable.
- is_all_free()¶
Indicates if all pool is free.
- async release(conn)¶
Release connection.
DNS Resolver¶
For custom dns servers, you sould install aiodns package and use Async resolver as follow
from aiosonic.resolver import AsyncResolver
resolver = AsyncResolver(nameservers=["8.8.8.8", "8.8.4.4"])
conn = aiosonic.TCPConnector(resolver=resolver)
Then, pass connector to aiosonic HTTPClient instance.
- class aiosonic.resolver.AsyncResolver(*args: Any, **kwargs: Any)¶
Use the aiodns package to make asynchronous DNS lookups
- async close() None ¶
Release resolver
- async resolve(host: str, port: int = 0, family: int = AddressFamily.AF_INET) List[Dict[str, Any]] ¶
Return IP address for given hostname
- class aiosonic.resolver.ThreadedResolver¶
Use Executor for synchronous getaddrinfo() calls, which defaults to concurrent.futures.ThreadPoolExecutor.
- async close() None ¶
Release resolver
- async resolve(hostname: str, port: int = 0, family: int = AddressFamily.AF_INET) List[Dict[str, Any]] ¶
Return IP address for given hostname
Proxy Support¶
- class aiosonic.proxy.Proxy(host: str, auth: Optional[str] = None)¶
Proxy class.
- Args:
host (str): proxy server where to connect
auth (str): auth data in the format of user:password