Reference¶

TODO: get better this page

Connector and Client Client¶

class aiosonic.connectors.TCPConnector(pool_size: int = 25, timeouts: Timeouts = None, connection_cls=None, pool_cls=None, resolver=None, ttl_dns_cache=10000, use_dns_cache=True, conn_max_requests=100)¶

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
conn_max_requests: Max requests allowed for a connection. default: 100

class aiosonic.HTTPClient(connector: TCPConnector | None = None, handle_cookies: bool = False, verify_ssl: bool = True, proxy: Proxy | None = 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.

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.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.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.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.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.SmartPool(connector, pool_size, connection_cls)¶

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.

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: 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.

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

Reference¶

Connector and Client Client¶

Classes¶

Tiemout Class¶

Pool Classes¶

DNS Resolver¶

Multipart Form Data¶

Proxy Support¶

aiosonic

Navigation

Related Topics