aiohttp: Async HTTP Client/Server

Install and use aiohttp, a powerful async HTTP client/server framework for Python built on asyncio. This skill covers making asynchronous HTTP requests, building web servers with middleware support, and handling WebSockets without callback hell.

What You'll Learn

Installation

Using pip

```bash

pip install aiohttp

```

Using uv

```bash

uv pip install aiohttp

```

Using conda

```bash

conda install -c conda-forge aiohttp

```

Optional (recommended for speed)

```bash

pip install aiodns

```

Instructions

1. Verify Installation

After installing, verify aiohttp is available:

```python

import aiohttp

print(aiohttp.__version__)

```

2. Making Async HTTP Requests (Client)

Use `ClientSession` to make async HTTP requests:

```python

import aiohttp

import asyncio

async def fetch_data():

async with aiohttp.ClientSession() as session:

async with session.get('http://python.org') as response:

print("Status:", response.status)

print("Content-type:", response.headers['content-type'])

html = await response.text()

print("Body:", html[:100], "...")

asyncio.run(fetch_data())

```

**Key patterns:**

3. Making POST Requests with JSON

Send JSON data to an API:

```python

import aiohttp

import asyncio

async def post_data():

async with aiohttp.ClientSession() as session:

data = {'key': 'value', 'number': 42}

async with session.post('https://httpbin.org/post', json=data) as response:

result = await response.json()

print("Response:", result)

asyncio.run(post_data())

```

4. Building a Simple Web Server

Create an async web server with routing:

```python

from aiohttp import web

async def hello_handler(request):

name = request.match_info.get('name', 'Anonymous')

return web.Response(text=f"Hello, {name}!")

async def json_handler(request):

data = {'message': 'Success', 'status': 'ok'}

return web.json_response(data)

app = web.Application()

app.add_routes([

web.get('/', hello_handler),

web.get('/api/data', json_handler),

web.get('/{name}', hello_handler)

])

if __name__ == '__main__':

web.run_app(app, port=8080)

```

Run the server and visit `http://localhost:8080/` or `http://localhost:8080/YourName`

5. Handling WebSockets

Implement a WebSocket echo server:

```python

from aiohttp import web

async def websocket_handler(request):

ws = web.WebSocketResponse()

await ws.prepare(request)

async for msg in ws:

if msg.type == web.WSMsgType.text:

await ws.send_str(f"Echo: {msg.data}")

elif msg.type == web.WSMsgType.binary:

await ws.send_bytes(msg.data)

elif msg.type == web.WSMsgType.close:

break

return ws

app = web.Application()

app.add_routes([web.get('/ws', websocket_handler)])

if __name__ == '__main__':

web.run_app(app, port=8080)

```

6. Using Middleware

Add logging or authentication middleware:

```python

from aiohttp import web

import time

@web.middleware

async def timing_middleware(request, handler):

start = time.time()

response = await handler(request)

elapsed = time.time() - start

print(f"{request.method} {request.path} took {elapsed:.2f}s")

return response

async def handler(request):

return web.Response(text="Hello!")

app = web.Application(middlewares=[timing_middleware])

app.add_routes([web.get('/', handler)])

if __name__ == '__main__':

web.run_app(app)

```

7. Handling POST Data (Forms)

Parse form data or multipart uploads:

```python

from aiohttp import web

async def upload_handler(request):

data = await request.post()

# Access form fields

username = data.get('username')

# Access uploaded files

file = data.get('file')

if file:

filename = file.filename

content = file.file.read()

print(f"Uploaded: {filename}, size: {len(content)} bytes")

return web.Response(text="Upload successful!")

app = web.Application()

app.add_routes([web.post('/upload', upload_handler)])

```

8. Session Timeout and Custom Headers

Configure client timeouts and headers:

```python

import aiohttp

import asyncio

async def fetch_with_config():

timeout = aiohttp.ClientTimeout(total=10) # 10 second timeout

headers = {

'User-Agent': 'MyApp/1.0',

'Authorization': 'Bearer token123'

}

async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session:

async with session.get('https://api.example.com/data') as response:

data = await response.json()

return data

asyncio.run(fetch_with_config())

```

Common Patterns

**Multiple concurrent requests:**

```python

async def fetch_all():

async with aiohttp.ClientSession() as session:

tasks = [

session.get('http://example.com/1'),

session.get('http://example.com/2'),

session.get('http://example.com/3')

]

responses = await asyncio.gather(*tasks)

return [await r.text() for r in responses]

```

**Error handling:**

```python

try:

async with session.get('http://example.com') as response:

response.raise_for_status() # Raises for 4xx/5xx

data = await response.json()

except aiohttp.ClientError as e:

print(f"Request failed: {e}")

```

Key Differences from `requests`

Resources

Important Notes