Examples

Download file

import asyncio
import aiosonic
import json


async def run():
    url = 'https://images.dog.ceo/breeds/leonberg/n02111129_2301.jpg'
    async with aiosonic.HTTPClient() as client:

       res = await client.get(url)
       assert res.status_code == 200

    if res.chunked:
        # write in chunks
        with open('dog_image.jpg', 'wb') as _file:
            async for chunk in res.read_chunks():
                _file.write(chunk)
    else:
        # or write all bytes, for chunked this also works
        with open('dog_image.jpg', 'wb') as _file:
            _file.write(await res.content())


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())

Concurrent Requests

import aiosonic
import asyncio


async def main():
    urls = [
        'https://www.facebook.com/',
        'https://www.google.com/',
        'https://twitch.tv/',
        'https://linkedin.com/',
    ]
    async with aiosonic.HTTPClient() as client:
        # asyncio.gather is the key for concurrent requests.
        responses = await asyncio.gather(*[client.get(url) for url in urls])
        assert all([res.status_code in [200, 301] for res in responses])

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Api Wrapping

import asyncio
import json
from aiosonic import BaseClient

class GitHubAPI(BaseClient):
    base_url = "https://api.github.com"
    default_headers = {
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
        # Uncomment the next line to use an authentication token
        # "Authorization": "Bearer YOUR_GITHUB_TOKEN",
    }

    async def process_response(self, response):
        body = await response.text()
        return json.loads(body)

    async def users(self, username: str, **kwargs):
        return await self.get(f"/users/{username}", **kwargs)


async def main():
    github = GitHubAPI()
    # Call the custom 'users' method to get data for user "sonic182"
    user_data = await github.users("sonic182")
    print(json.dumps(user_data, indent=2))


if __name__ == '__main__':
    asyncio.run(main())

This example demonstrates how to use BaseClient to define a specialized API client. The users() method hides the details of performing the underlying HTTP request so that your application code can remain clean and focused on the API semantics.

Chunked Requests (Stream request or response)

Specifying an iterator as the request body, it will make the request transfer made by chunks

import aiosonic
import asyncio
import json


async def main():
    async def data():
        yield b'foo'
        yield b'bar'

    async with aiosonic.HTTPClient() as client:
        url = 'https://postman-echo.com/post'
        response = await client.post(url, data=data())
        print(json.dumps(await response.json(), indent=10))


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Multiple Pool Configurations

You can configure different connection pools for different domains, which is useful when you need specialized settings for specific services.

import aiosonic
import asyncio
from aiosonic.pools import PoolConfig


async def main():
    pool_configs = {
        "https://www.google.com": PoolConfig(
            size=5,  # Only 5 connections for Google
            max_conn_requests=100  # Recycle connection after 100 requests
        ),
        "https://api.github.com": PoolConfig(
            size=20,  # More connections for GitHub API
            max_conn_idle_ms=60000  # Close idle connections after 60 seconds
        ),
        ":default": PoolConfig(
            size=30  # Use 30 connections for any other domains
        )
    }

    # Create connector with custom pool configurations
    connector = aiosonic.TCPConnector(pool_configs=pool_configs)

    async with aiosonic.HTTPClient(connector=connector) as client:
        # ... client usage
        pass

Cookies handling

Adding handle_cookies=True to the client, it will save response cookies and send it again for new requests. This is useful to have same cookies workflow as in browsers, also for web scraping.

import aiosonic
import asyncio


async def main():
    async with aiosonic.HTTPClient(handle_cookies=True) as client:
        cookies = {'foo1': 'bar1', 'foo2': 'bar2'}
        url = 'https://postman-echo.com/cookies/set'
        # server will respond those cookies
        response = await client.get(url, params=cookies, follow=True)
        # client keep cookies in "cookies_map"
        print(client.cookies_map['postman-echo.com'])
        print(await response.text())


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Use custom DNS

Install aiodns in your dependencies and use AsyncResolver

import aiosonic
import asyncio
from aiosonic.resolver import AsyncResolver


async def main():
    resolver = AsyncResolver(nameservers=["8.8.8.8", "8.8.4.4"])
    connector = aiosonic.TCPConnector(resolver=resolver)

    async with aiosonic.HTTPClient(connector=connector) as client:
        data = {'foo1': 'bar1', 'foo2': 'bar2'}
        url = 'https://postman-echo.com/post'
        # server will respond those cookies
        response = await client.post(url, json=data)
        # client keep cookies in "cookies_map"
        print(await response.text())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Use a Proxy Server

Just use Proxy class.

You can install proxy.py and use it as a proxy demo.

import asyncio

from aiosonic import HTTPClient, Proxy


async def main():
    # Proxy class accepts `auth` argument in the format `user:password`
    client = HTTPClient(proxy=Proxy("http://localhost:8899"))

    res = await client.get("https://www.google.com/")
    print(res)
    print(await res.text())
    assert res.status_code == 200


asyncio.run(main())

Debug log

Configure aiosonic logger at debug level to see some logging

import asyncio
import aiosonic
import json
import logging


async def run():
    # setup debug level at log
    logger = logging.getLogger('aiosonic')
    logger.setLevel(logging.DEBUG)

    async with aiosonic.HTTPClient() as client:
      response = await client.get('https://www.google.com/')
      assert response.status_code == 200
      assert 'Google' in (await response.text())

loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Fastapi Usage

from contextlib import asynccontextmanager
from typing import Union

from aiosonic import HTTPClient

from fastapi import FastAPI

client = HTTPClient()


@asynccontextmanager
async def lifespan(_app: FastAPI):
    global client
    yield
    # useful function to wait all pending requests to finish
    await client.wait_requests()


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def home():
    assert client, "no client"
    url = "https://postman-echo.com/post"
    res = await client.post(url, params={"foo": "bar"})
    return (await res.json())["args"]