Reference

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, http2: bool = False)

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, or Http2MultiplexPool when http2=True.

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.

http2 (bool):

When True and pool_cls is not explicitly provided, defaults pool_cls to Http2MultiplexPool. Defaults to False.


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

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.

  • http2: Flag to enable HTTP/2 for all requests made by this client.

    Per-request http2=True also works and takes precedence.


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 json and appends 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.

property http_version: str

Get the negotiated HTTP version string (e.g. ‘2’, ‘1.1’, ‘1.0’).

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.


SSE Client

class aiosonic.SSEClient(http_client: HTTPClient | None = None)

Bases: object

SSE client for aiosonic.

Examples:

Basic usage with async context manager:

>>> import asyncio
>>> from aiosonic import SSEClient
>>>
>>> async def main():
...     async with SSEClient() as client:
...         async with client.connect("http://example.com/sse") as sse_conn:
...             async for event in sse_conn:
...                 print(f"Event: {event['event']}, Data: {event['data']}")
...                 if event['data'] == 'stop':
...                     break
>>>
>>> asyncio.run(main())

POST request with JSON body (OpenAI-style streaming):

>>> async def main():
...     async with SSEClient() as client:
...         async with client.connect(
...             "https://api.openai.com/v1/chat/completions",
...             method="POST",
...             json={"model": "gpt-4", "messages": [...], "stream": True},
...             headers={"Authorization": "Bearer token"}
...         ) as sse_conn:
...             async for event in sse_conn:
...                 print(event['data'])
>>>
>>> asyncio.run(main())
connect(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, multipart: bool = False, verify: bool = True, ssl: ~ssl.SSLContext | None = None, timeouts: ~aiosonic.timeout.Timeouts | None = None, follow: bool = False, http2: bool = False, reconnect: bool = True, retry_delay: int = 3000, json_serializer=<function dumps>, keep_connection: bool = False) → _ConnectAwaitable

Return an awaitable/async-context-manager that establishes an SSE connection.

Args:

url: The URL to connect to method: HTTP method to use (GET, POST, PUT, PATCH, DELETE, etc.) headers: HTTP headers to send params: Query parameters to include in the URL data: Request body data json: JSON data to send as request body multipart: Whether to send data as multipart form verify: Whether to verify SSL certificates ssl: Custom SSL context timeouts: Request timeout settings follow: Whether to follow redirects http2: Whether to use HTTP/2 reconnect: Whether to automatically reconnect on connection loss retry_delay: Delay between reconnection attempts in milliseconds json_serializer: Custom JSON serializer function keep_connection: Whether to keep the connection open after the SSE stream ends

(experimental)

Usage:

  • sse_conn = await client.connect(url, method=”POST”, json=data)

  • async with client.connect(url, method=”POST”, json=data) as sse_conn:

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 = 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 = 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, MultipartFile

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

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

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

    # Add a file using MultipartFile for more control (with file path)
    multipart_file = MultipartFile("img.png", filename="custom.png", content_type="image/png")
    form.add_field("image", multipart_file)

    # Or with an already opened file object (caller responsible for closing)
    file_obj = open("path/to/your/document.pdf", "rb")
    multipart_file2 = MultipartFile(file_obj, filename="doc.pdf", content_type="application/pdf")
    form.add_field("document", multipart_file2)
    # Close the file after the request if needed
    # file_obj.close()

    # 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 | MultipartFile, filename: str | None = None)

Adds a field to the multipart form data.

Args:

name (str): The name of the field. value (Union[str, IOBase, MultipartFile]): Field value: str, IOBase, or MultipartFile. filename (Optional[str]): File name if value is file-like or MultipartFile.

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]): File name for server. Defaults to file’s name.

async get_body_size()

Calculate total multipart body size and return with body.

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: (multipart body bytes, size).

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.


class aiosonic.multipart.MultipartFile(file_path_or_obj: str | IOBase, filename: str | None = None, content_type: str | None = None)

A class to represent a file in multipart data with metadata.

This class encapsulates a file path or file object along with its filename and content type, providing convenient access to file properties such as size.

Args:

file_path_or_obj (Union[str, IOBase]): Path to file or opened file object for multipart data. filename (Optional[str]): The name of the file. If not provided, defaults to

the basename of the file path or file object’s name.

content_type (Optional[str]): The MIME type of the file. If not provided,

it may be inferred or left unspecified.

property file_obj: IOBase

Return the file object, opening it if necessary.

property size: int

Calculate and cache the file size efficiently.

Proxy Support

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

Proxy class.

Args:

  • host (str): proxy server where to connect

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