Reference

TODO: get better this page

Connector and Client Client

class aiosonic.connectors.TCPConnector(pool_configs: Dict[str, PoolConfig | Dict[str, Any]] | None = None, timeouts: Timeouts | None = 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.

Parameters:
pool_configs (Optional[Dict[str, PoolConfig]]):

Map of host domains to pool configurations. Keys are host domains (e.g., “https://example.com” or “example.com”) and values are PoolConfig instances. A special key “:default” is used for hosts without a specific configuration.

timeouts (Optional[Timeouts]):

Global timeouts for connections. Defaults to a Timeouts instance with default args.

connection_cls:

Connection class to be used. Defaults to Connection.

pool_cls:

Pool class to be used. Defaults to SmartPool.

resolver:

DNS resolver to be used. Defaults to DefaultResolver.

ttl_dns_cache (int):

TTL in milliseconds for DNS cache. Defaults to 10000 (10 seconds).

use_dns_cache (bool):

Flag to indicate usage of DNS cache. Defaults to True.


class aiosonic.HTTPClient(connector: TCPConnector | None = None, handle_cookies: bool = False, verify_ssl: bool = True, proxy: Proxy | None = None, max_redirects: int = 5)

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: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | ~aiosonic.client.HttpHeaders | None = None, params: ~typing.Dict[str, str] | ~typing.Sequence[~typing.Tuple[str, str]] | None = None, data: str | bytes | dict | tuple | ~typing.AsyncIterator[bytes] | ~typing.Iterator[bytes] | ~aiosonic.multipart.MultipartForm | None = None, json: dict | list | None = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: ~ssl.SSLContext | None = None, timeouts: ~aiosonic.timeout.Timeouts | None = None, follow: bool = False, http2: bool = False, max_redirects: int | None = None) → 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

  • json: If provided, encodes the provided json structure and appends the corresponding header.

  • json_serializer: Use provided json serializer, default: json.dumps

  • 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: Dict[str, str] | List[Tuple[str, str]] | HttpHeaders | None = None, params: Dict[str, str] | Sequence[Tuple[str, str]] | None = None, verify: bool = True, ssl: SSLContext | None = None, timeouts: Timeouts | None = None, follow: bool = False, http2: bool = False) → HttpResponse

Do get http request.


async aiosonic.HTTPClient.post(self, url: str, data: str | bytes | dict | tuple | ~typing.AsyncIterator[bytes] | ~typing.Iterator[bytes] | ~aiosonic.multipart.MultipartForm | None = None, headers: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | ~aiosonic.client.HttpHeaders | None = None, json: dict | list | None = None, params: ~typing.Dict[str, str] | ~typing.Sequence[~typing.Tuple[str, str]] | None = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: ~ssl.SSLContext | None = None, timeouts: ~aiosonic.timeout.Timeouts | None = None, follow: bool = False, http2: bool = False) → HttpResponse

Do post http request.


async aiosonic.HTTPClient.put(self, url: str, data: str | bytes | dict | tuple | ~typing.AsyncIterator[bytes] | ~typing.Iterator[bytes] | ~aiosonic.multipart.MultipartForm | None = None, headers: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | ~aiosonic.client.HttpHeaders | None = None, json: dict | list | None = None, params: ~typing.Dict[str, str] | ~typing.Sequence[~typing.Tuple[str, str]] | None = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: ~ssl.SSLContext | None = None, timeouts: ~aiosonic.timeout.Timeouts | None = None, follow: bool = False, http2: bool = False) → HttpResponse

Do put http request.


async aiosonic.HTTPClient.patch(self, url: str, data: str | bytes | dict | tuple | ~typing.AsyncIterator[bytes] | ~typing.Iterator[bytes] | ~aiosonic.multipart.MultipartForm | None = None, headers: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | ~aiosonic.client.HttpHeaders | None = None, json: dict | list | None = None, params: ~typing.Dict[str, str] | ~typing.Sequence[~typing.Tuple[str, str]] | None = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: ~ssl.SSLContext | None = None, timeouts: ~aiosonic.timeout.Timeouts | None = None, follow: bool = False, http2: bool = False) → HttpResponse

Do patch http request.


async aiosonic.HTTPClient.delete(self, url: str, data: str | bytes | dict | tuple | ~typing.AsyncIterator[bytes] | ~typing.Iterator[bytes] | ~aiosonic.multipart.MultipartForm = b'', headers: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | ~aiosonic.client.HttpHeaders | None = None, json: dict | list | None = None, params: ~typing.Dict[str, str] | ~typing.Sequence[~typing.Tuple[str, str]] | None = None, json_serializer=<function dumps>, multipart: bool = False, verify: bool = True, ssl: ~ssl.SSLContext | None = None, timeouts: ~aiosonic.timeout.Timeouts | None = None, follow: bool = False, http2: bool = False) → 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.

property ok: bool

Returns True if status_code is 2xx range, False if not.

async 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: float | None = 5, sock_read: float | None = 30, pool_acquire: float | None = None, request_timeout: float | None = 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.PoolConfig(size: int = 30, max_conn_requests: int | None = 1000, max_conn_idle_ms: int = 60000)

Configuration options for database connection pools.

Controls how connections are created, maintained, and recycled.

This class is immutable and hashable, allowing it to be used as a dictionary key.

class aiosonic.pools.SmartPool(conf: PoolConfig, connection_cls, timeouts: Timeouts | None = None)

Pool which priorizes the reusage of connections.

async acquire(urlparsed: ParseResult = None)

Acquire connection.

async cleanup() → None

Get all conn and close them, this method let this pool unusable.

free_conns() → int

Return number of free connections.

is_all_free()

Indicates if all pool is free.

release(conn) → None

Release connection.


class aiosonic.pools.CyclicQueuePool(conf: PoolConfig, connection_cls, timeouts: Timeouts | None = None)

Cyclic queue pool of connections.

async acquire(_urlparsed: ParseResult = None)

Acquire connection.

async cleanup()

Get all conn and close them, this method let this pool unusable.

free_conns() → int

Return number of free connections.

is_all_free()

Indicates if all pool is free.

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

Multipart Form Data

This class can be used for sending multipart form data.

class aiosonic.multipart.MultipartForm

A class to handle multipart form data for HTTP requests.

Example:

import asyncio
import aiosonic
from aiosonic.multipart import MultipartForm

async def upload_file():
    client = aiosonic.HTTPClient()
    form = MultipartForm()

    # Add a text field
    form.add_field("field1", "value1")

    # Add a file to upload
    form.add_file("file1", "path/to/your/file.txt")

    # Make the POST request with MultipartForm directly
    url = "https://your-upload-endpoint.com/upload"
    response = await client.post(url, data=form)

    print("Response Status:", response.status_code)
    response_data = await response.text()
    print("Response Body:", response_data)

if __name__ == '__main__':
    asyncio.run(upload_file())
add_field(name: str, value: str | IOBase, filename: str | None = None)

Adds a field to the multipart form data.

Args:

name (str): The name of the field. value (Union[str, IOBase]): The value of the field. Can be a string or a file-like object. filename (Optional[str]): The name of the file, if the value is a file-like object.

Defaults to the file’s name if not provided.

add_file(name: str, file_path: str, filename: str | None = None)

Adds a file to the multipart form data.

The file is opened and it is closed after the request is sent.

Args:

name (str): The name of the file field. file_path (str): The file path of the file to be added. filename (Optional[str]): The name of the file for the target server. defaults to the file’s name.

async get_body_size()

Calculates the total size of the multipart body and returns it along with the body itself.

This function asynchronously constructs the body by iterating over the chunks generated by the get_buffer method. It accumulates the total size of the body in bytes while building the complete body as a byte string.

Returns:

tuple: A tuple containing the complete multipart body as bytes and its size in bytes.

async get_buffer()

Returns an asynchronous iterator that generates the constructed multipart buffer.

get_headers(size=None)

Returns the headers for the multipart form data.

Proxy Support

class aiosonic.proxy.Proxy(host: str, auth: str = None)

Proxy class.

Args:

  • host (str): proxy server where to connect

  • auth (str): auth data in the format of user:password