Async http client/server framework (asyncio)
npx @tessl/cli install tessl/pypi-aiohttp@3.12.0A comprehensive asynchronous HTTP client/server framework built on Python's asyncio. aiohttp supports both client and server side of HTTP protocol with built-in WebSocket support, avoiding callback hell through async/await syntax. The framework offers a web server with middleware and pluggable routing capabilities, making it suitable for building modern web applications and APIs.
pip install aiohttpimport aiohttpFor HTTP client operations:
import aiohttp
# or import specific components
from aiohttp import ClientSession, ClientTimeoutFor web server development:
from aiohttp import web
# or import specific web components
from aiohttp.web import Application, Response, json_response, run_appComplete web framework import:
# Import the web framework
from aiohttp import web
# Access all web components through web module
app = web.Application()
response = web.Response(text="Hello World")import aiohttp
import asyncio
async def fetch_data():
async with aiohttp.ClientSession() as session:
async with session.get('https://api.example.com/data') as response:
data = await response.json()
return data
# Run the async function
data = asyncio.run(fetch_data())from aiohttp import web
async def hello_handler(request):
return web.Response(text="Hello, World!")
app = web.Application()
app.router.add_get('/', hello_handler)
if __name__ == '__main__':
web.run_app(app, host='localhost', port=8080)aiohttp follows a layered architecture:
This design enables both high-performance HTTP clients and scalable web servers while maintaining full asyncio compatibility.
Comprehensive HTTP client functionality including sessions, connectors, authentication, and connection management. Supports all HTTP methods, automatic redirects, cookie handling, and extensive configuration options.
class ClientSession:
def __init__(self, timeout=None, connector=None, cookie_jar=None, **kwargs): ...
async def request(method, url, **kwargs): ...
class ClientResponse:
async def text(self): ...
async def json(self): ...
async def read(self): ...Full-featured web server with applications, routing, middleware, and request/response handling. Includes support for static files, WebSocket connections, and extensible middleware system.
class Application:
def __init__(self, **kwargs): ...
class Request:
async def text(self): ...
async def json(self): ...
async def post(self): ...
class Response:
def __init__(self, text=None, body=None, status=200, **kwargs): ...
def run_app(app, host=None, port=None, **kwargs): ...Client and server WebSocket implementation with message handling, connection management, and protocol support. Enables real-time bidirectional communication.
class ClientWebSocketResponse:
async def send_str(self, data): ...
async def receive(self): ...
async def close(self): ...
class WebSocketResponse:
async def prepare(self, request): ...
async def send_str(self, data): ...
async def receive(self): ...Comprehensive data processing system supporting various payload types, multipart handling, form data, and streaming. Includes automatic content encoding and decoding.
class FormData:
def add_field(self, name, value, **kwargs): ...
class MultipartReader:
async def next(self): ...
class MultipartWriter:
append(self, obj, headers=None): ...Complete testing utilities including test servers, test clients, and pytest integration. Enables comprehensive testing of both client and server applications.
class TestClient:
async def get(self, path, **kwargs): ...
async def post(self, path, **kwargs): ...
class TestServer:
def __init__(self, app, **kwargs): ...Supporting utilities including tracing, DNS resolution, authentication helpers, and Gunicorn workers. Provides integration points and extensibility features.
class TraceConfig:
def on_request_start(self, callback): ...
def on_request_end(self, callback): ...
class BasicAuth:
def __init__(self, login, password): ...import json
from typing import Optional, Union, Dict, Any, Callable
from multidict import CIMultiDict
from yarl import URL
# Type aliases used throughout the API
StrOrURL = Union[str, URL]
LooseHeaders = Union[Dict[str, str], CIMultiDict]
JsonSerializable = Union[Dict[str, Any], list, str, int, float, bool, None]class ClientTimeout:
def __init__(self, total=None, sock_connect=None, sock_read=None): ...
class RequestInfo:
url: URL
method: str
headers: CIMultiDictclass AppKey:
def __init__(self, name: str, t: type): ...
Handler = Callable[[Request], Response]
Middleware = Callable[[Request, Handler], Response]