Install and use aiohttp, the async HTTP client/server framework for Python. Learn to make async requests, build web servers, and handle WebSockets with asyncio.
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.
```bash
pip install aiohttp
```
```bash
uv pip install aiohttp
```
```bash
conda install -c conda-forge aiohttp
```
```bash
pip install aiodns
```
After installing, verify aiohttp is available:
```python
import aiohttp
print(aiohttp.__version__)
```
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:**
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())
```
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`
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)
```
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)
```
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)])
```
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())
```
**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}")
```
Leave a review
No reviews yet. Be the first to review this skill!
# Download SKILL.md from killerskills.ai/api/skills/aiohttp-async-http-clientserver/raw