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)

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.

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.

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