aioquic

Warnings

• breaking As of version 1.3.0, aioquic has dropped support for Python 3.8 and 3.9. Users must use Python 3.10 or newer. Newer versions also include support for Python 3.13 and 3.14.
  Fix: Upgrade your Python environment to version 3.10 or higher.
• gotcha When building aioquic from source (e.g., if pre-built wheels are not available for your platform), you must have OpenSSL development headers installed on your system. This is a common non-Python dependency that can cause installation failures.
  Fix: Install OpenSSL development headers for your operating system (e.g., `sudo apt install libssl-dev python3-dev` on Debian/Ubuntu, `brew install openssl` on macOS).
• gotcha The core QUIC and HTTP/3 APIs in aioquic follow a 'sans-I/O' pattern, meaning they do not perform network I/O directly. Users are responsible for sending and receiving UDP datagrams. While the `aioquic.asyncio` API provides convenience, direct interaction with `QuicConnection` requires manual datagram handling, which can be a point of confusion for new users.
  Fix: For simplified I/O, utilize the `aioquic.asyncio` module and its `QuicConnectionProtocol`, `connect`, and `serve` functions. When using the lower-level `QuicConnection` directly, ensure you regularly call `datagrams_to_send()` and handle `receive_datagram()` based on network events.
• gotcha Running an aioquic server, especially for HTTP/3, requires a TLS certificate and private key. For development or testing, self-signed certificates can be generated, but their presence and correct configuration in `QuicConfiguration` are mandatory.
  Fix: Generate a self-signed certificate and key (e.g., using OpenSSL: `openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365`) and configure `QuicConfiguration` with `certificatefile` and `private_key_file`.

Install

• pip install aioquic Install stable version

Imports

• QuicConnectionProtocol

  from aioquic.asyncio import QuicConnectionProtocol

  Commonly used in asyncio-based client and server implementations.
• connect

  from aioquic.asyncio import connect

  Used by asyncio clients to initiate a QUIC connection.
• serve

  from aioquic.asyncio import serve

  Used by asyncio servers to listen for incoming QUIC connections.
• QuicConfiguration

  from aioquic.quic.configuration import QuicConfiguration

  Essential for configuring QUIC parameters like ALPN protocols, TLS certificates, and keys.
• H3_ALPN

  from aioquic.h3.connection import H3_ALPN

  Specifies the Application-Layer Protocol Negotiation (ALPN) token for HTTP/3.
• H3Connection

  from aioquic.h3.connection import H3Connection

  Represents the HTTP/3 layer for sending and receiving HTTP/3 events.
• HeadersReceived

  from aioquic.h3.events import HeadersReceived

  Event type representing received HTTP/3 headers.
• DataReceived

  from aioquic.h3.events import DataReceived

  Event type representing received HTTP/3 data.

Quickstart

This quickstart demonstrates a basic HTTP/3 client using `aioquic` to connect to a server and fetch content. It initializes a `QuicConfiguration` for client-side operation with HTTP/3 ALPN, then connects to a specified host and port. The `HttpClientProtocol` handles QUIC events, processes HTTP/3 headers and data, and prints the response body. This example showcases how to establish a connection and make a simple GET request.

import asyncio
import logging
from typing import Optional

from aioquic.asyncio import connect
from aioquic.asyncio.protocol import QuicConnectionProtocol
from aioquic.h3.connection import H3_ALPN, H3Connection
from aioquic.h3.events import DataReceived, HeadersReceived, H3Event
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import QuicEvent

class HttpClientProtocol(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._http = H3Connection(self._quic)
        self._buffer = b''
        self._http_events = asyncio.Queue()
        self._task = asyncio.create_task(self._http_event_handler())

    async def _http_event_handler(self):
        while True:
            event = await self._http_events.get()
            if isinstance(event, HeadersReceived):
                for k, v in event.headers:
                    print(f"Header: {k.decode()} = {v.decode()}")
            elif isinstance(event, DataReceived):
                self._buffer += event.data
                if event.stream_ended:
                    print(f"Body: {self._buffer.decode()}")
                    self._buffer = b''

    def quic_event_received(self, event: QuicEvent) -> None:
        self._http.handle_event(event)
        while True:
            http_event = self._http.pull_http_event()
            if http_event is None:
                break
            self._http_events.put_nowait(http_event)

    async def get(self, url: str):
        headers = [
            (b":method", b"GET"),
            (b":scheme", b"https"),
            (b":authority", url.encode()),
            (b":path", b"/"),
            (b"user-agent", b"aioquic-client/1.0"),
        ]
        stream_id = self._http.send_headers(stream_id=self._http.get_next_available_stream_id(), headers=headers)
        self._http.send_data(stream_id=stream_id, data=b'', end_stream=True)
        self.transmit()

async def main():
    logging.basicConfig(level=logging.INFO)

    configuration = QuicConfiguration(
        is_client=True,
        alpn_protocols=H3_ALPN,
    )
    # For testing against a local server with a self-signed cert, you might need to disable certificate verification
    # configuration.verify_mode = ssl.CERT_NONE

    host = "quic.aiortc.org"
    port = 4433

    async with connect(
        host=host,
        port=port,
        configuration=configuration,
        create_protocol=HttpClientProtocol,
    ) as client_protocol:
        await client_protocol.get(host)
        await asyncio.sleep(1) # Give time for events to be processed

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