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 dict

  • cookies (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