Welcome to aiosonic

A really fast, lightweight Python asyncio HTTP/1.1, HTTP/2, and WebSocket client.

Current version is 0.27.0.

The repository is hosted on GitHub:

Features

  • Keepalive support and a smart pool of connections

  • Multipart file uploads

  • Handling of chunked responses and requests

  • Connection timeouts and automatic decompression

  • Automatic redirect following

  • Fully type-annotated code

  • WebSocket support

  • Server-Sent Events (SSE) support

  • HTTP proxy support

  • Sessions with cookie persistence

  • Elegant key/value cookies

  • Comprehensive test coverage (nearly 100%)

  • HTTP/2 (BETA; enabled via a flag)

Requirements

  • Python >= 3.8 (or PyPy 3.8+)

Installation

$ pip install aiosonic

Getting Started

Below is a basic example of using aiosonic’s HTTP client:

import asyncio
import aiosonic
import json

async def run():
    client = aiosonic.HTTPClient()

    # Sample GET request
    response = await client.get('https://www.google.com/')
    assert response.status_code == 200
    assert 'Google' in (await response.text())

    # POST data as multipart form
    url = "https://postman-echo.com/post"
    posted_data = {'foo': 'bar'}
    response = await client.post(url, data=posted_data)
    assert response.status_code == 200
    data = json.loads(await response.content())
    assert data['form'] == posted_data

    # POST data as JSON
    response = await client.post(url, json=posted_data)
    assert response.status_code == 200
    data = json.loads(await response.content())
    assert data['json'] == posted_data

    # GET request with timeouts
    from aiosonic.timeout import Timeouts
    timeouts = Timeouts(sock_read=10, sock_connect=3)
    response = await client.get('https://www.google.com/', timeouts=timeouts)
    assert response.status_code == 200
    assert 'Google' in (await response.text())

    print('HTTP client success')

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

WebSocket Example

This example demonstrates how to use the WebSocket support provided by aiosonic.

import asyncio
from aiosonic import WebSocketClient

async def main():
    # Replace with your WebSocket server URL
    ws_url = "ws://localhost:8080"
    async with WebSocketClient() as client:
        async with await client.connect(ws_url) as ws:
            # Send a text message.
            await ws.send_text("Hello WebSocket")

            # Receive the echo response.
            response = await ws.receive_text()
            print("Received:", response)

            # Send a ping and wait for the pong response.
            await ws.ping(b"keep-alive")
            pong = await ws.receive_pong()
            print("Pong received:", pong)

            # Gracefully close the connection.
            await ws.close(code=1000, reason="Normal closure")

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

SSE Example

This example demonstrates how to use the Server-Sent Events (SSE) support provided by aiosonic.

import asyncio
from aiosonic import SSEClient

async def main():
    # Replace with your SSE endpoint URL
    sse_url = "http://localhost:8080/sse"
    client = SSEClient()
    async with client.connect(sse_url) as sse_conn:
        async for event in sse_conn:
            print(f"Event: {event['event']}, Data: {event['data']}")
            if event['data'] == 'stop':
                break

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

Benchmarks

Below is a basic performance benchmark comparing aiosonic with other HTTP clients:

$ python scripts/performance.py
{
  "aiohttp": "5000 requests in 558.31 ms",
  "aiosonic": "5000 requests in 563.95 ms",
  "requests": "5000 requests in 10306.90 ms",
  "aiosonic_cyclic": "5000 requests in 642.15 ms",
  "httpx": "5000 requests in 7920.04 ms"
}

Note that these benchmarks are machine-dependent and intended only as a rough comparison.

Contributing

  1. Fork the repository.

  2. Create a branch (e.g. feature/your_feature).

  3. Commit, push, and submit a pull request.

Thanks to all contributors!

Indices and Tables