Логирование, CORS, rate limiting, кастомные middleware
Middleware обрабатывает запросы до и после endpoints. В этой теме вы научитесь логированию, CORS, rate limiting и созданию кастомных middleware.
Middleware — функция, выполняющаяся до и после обработки запроса endpoint'ом.
Запрос → Middleware 1 → Middleware 2 → Endpoint → Response → Middleware 2 → Middleware 1 → Клиент
| Задача | Middleware | Почему |
|---|---|---|
| Логирование запросов | ✅ | Все запросы проходят через middleware |
| CORS | ✅ | Добавляет заголовки ко всем ответам |
| Rate limiting | ✅ | Проверка до выполнения endpoint |
| Аутентификация | ✅ | Проверка токена до доступа |
| Сжатие ответов | ✅ | Обработка всех ответов |
| Валидация данных | ❌ | Делается в endpoint через Pydantic |
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next):
"""
Логирование всех запросов.
"""
# До выполнения
start_time = time.time()
print(f"{request.method} {request.url.path}")
# Выполнение запроса
response = await call_next(request)
# После выполнения
duration = time.time() - start_time
print(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s")
return responsefrom starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
class CustomHeaderMiddleware(BaseHTTPMiddleware):
"""
Добавляет кастомные заголовки ко всем ответам.
"""
async def dispatch(self, request: Request, call_next):
# До выполнения
request.state.custom_data = "some data"
# Выполнение
response = await call_next(request)
# После выполнения
response.headers['X-Custom-Header'] = 'Custom Value'
response.headers['X-Request-ID'] = request.headers.get('X-Request-ID', 'unknown')
return response
app.add_middleware(CustomHeaderMiddleware)CORS (Cross-Origin Resource Sharing) — разрешение запросов с других доменов.
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# Настройка CORS
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000", # React
"http://localhost:8080", # Vue
"https://myapp.com", # Production
],
allow_credentials=True, # Разрешить cookies
allow_methods=["*"], # Все методы (GET, POST, PUT, DELETE)
allow_headers=["*"], # Все заголовки
)| Параметр | Значение | Описание |
|---|---|---|
allow_origins | ["https://site.com"] | Разрешённые домены |
allow_credentials | True/False | Разрешить cookies/authorization |
allow_methods | ["GET", "POST"] | Разрешённые методы |
allow_headers | ["*"] | Разрешённые заголовки |
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Все домены
allow_credentials=False, # Cookies не работают с "*"
allow_methods=["*"],
allow_headers=["*"],
)Ограничение количества запросов от одного клиента.
from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import JSONResponse
from collections import defaultdict
import time
app = FastAPI()
# Хранилище: {ip: [(timestamp, count)]}
request_history = defaultdict(list)
RATE_LIMIT = 10 # запросов
TIME_WINDOW = 60 # секунд
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
client_ip = request.client.host
current_time = time.time()
# Очистка старых записей
request_history[client_ip] = [
(ts, count) for ts, count in request_history[client_ip]
if current_time - ts < TIME_WINDOW
]
# Подсчёт запросов
total_requests = sum(count for _, count in request_history[client_ip])
if total_requests >= RATE_LIMIT:
return JSONResponse(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
content={"detail": "Rate limit exceeded. Try again later."}
)
# Добавление текущего запроса
request_history[client_ip].append((current_time, 1))
# Выполнение запроса
response = await call_next(request)
# Добавление заголовков
response.headers['X-RateLimit-Limit'] = str(RATE_LIMIT)
response.headers['X-RateLimit-Remaining'] = str(RATE_LIMIT - total_requests - 1)
return responsefrom fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import redis
import time
app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379)
RATE_LIMIT = 100 # запросов в минуту
TIME_WINDOW = 60 # секунд
@app.middleware("http")
async def redis_rate_limit(request: Request, call_next):
client_ip = request.client.host
key = f"rate_limit:{client_ip}"
# Получение текущего счёта
current = redis_client.get(key)
if current and int(current) >= RATE_LIMIT:
ttl = redis_client.ttl(key)
return JSONResponse(
status_code=429,
content={
"detail": "Rate limit exceeded",
"retry_after": ttl
},
headers={"Retry-After": str(ttl)}
)
# Инкремент счёта
pipe = redis_client.pipeline()
pipe.incr(key)
pipe.expire(key, TIME_WINDOW)
pipe.execute()
# Выполнение запроса
response = await call_next(request)
# Заголовки
remaining = RATE_LIMIT - int(redis_client.get(key) or 0)
response.headers['X-RateLimit-Limit'] = str(RATE_LIMIT)
response.headers['X-RateLimit-Remaining'] = str(max(0, remaining))
return responseСжатие ответов для уменьшения трафика.
from fastapi.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000) # Сжимать от 1KB| Параметр | Значение | Описание |
|---|---|---|
minimum_size | 1000 | Минимальный размер для сжатия (байты) |
compress_level | 5 | Уровень сжатия (1-9, по умолчанию 5) |
Перенаправление с HTTP на HTTPS.
from fastapi import FastAPI, Request
from starlette.responses import RedirectResponse
app = FastAPI()
@app.middleware("http")
async def https_redirect(request: Request, call_next):
# Только для production
if request.url.scheme == "http" and request.url.hostname != "localhost":
url = request.url.replace(scheme="https")
return RedirectResponse(url, status_code=301)
return await call_next(request)Добавление заголовков безопасности.
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def security_headers(request: Request, call_next):
response = await call_next(request)
# Защита от XSS
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-XSS-Protection'] = '1; mode=block'
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
# Content Security Policy
response.headers['Content-Security-Policy'] = (
"default-src 'self'; "
"script-src 'self'; "
"style-src 'self' 'unsafe-inline'"
)
return responseДобавление заголовка с временем выполнения.
from fastapi import FastAPI, Request
import time
app = FastAPI()
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
response.headers['X-Process-Time'] = str(duration)
return responsefrom fastapi import FastAPI, Request, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging
import redis
app = FastAPI()
# Логирование
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Redis для rate limiting
redis_client = redis.Redis(host='localhost', port=6379)
# === Middleware ===
# 1. CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["https://myapp.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 2. GZip
app.add_middleware(GZipMiddleware, minimum_size=1000)
# 3. Логирование
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
logger.info(f"{request.method} {request.url.path}")
response = await call_next(request)
duration = time.time() - start_time
logger.info(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s")
response.headers['X-Process-Time'] = str(duration)
return response
# 4. Rate Limiting
RATE_LIMIT = 100
TIME_WINDOW = 60
@app.middleware("http")
async def rate_limit(request: Request, call_next):
client_ip = request.client.host
key = f"rate:{client_ip}"
current = redis_client.get(key)
if current and int(current) >= RATE_LIMIT:
raise HTTPException(
status_code=429,
detail="Too many requests",
headers={"Retry-After": str(redis_client.ttl(key))}
)
pipe = redis_client.pipeline()
pipe.incr(key)
pipe.expire(key, TIME_WINDOW)
pipe.execute()
response = await call_next(request)
remaining = RATE_LIMIT - int(redis_client.get(key) or 0)
response.headers['X-RateLimit-Remaining'] = str(remaining)
return response
# 5. Security Headers
@app.middleware("http")
async def security_headers(request: Request, call_next):
response = await call_next(request)
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-XSS-Protection'] = '1; mode=block'
return response
# === Endpoints ===
@app.get('/')
def read_root():
return {'message': 'Hello World'}
@app.get('/slow')
def slow_endpoint():
time.sleep(2)
return {'message': 'Slow response'}
@app.get('/large')
def large_endpoint():
# Большой ответ (будет сжат GZip)
return {'data': 'x' * 10000}Middleware выполняются в порядке добавления:
app.add_middleware(Middleware1) # Выполняется первым ДО, последним ПОСЛЕ
app.add_middleware(Middleware2) # Выполняется вторым ДО, первым ПОСЛЕ
app.add_middleware(Middleware3) # Выполняется последним ДО, вторым ПОСЛЕ
@app.middleware("http")
def middleware4(): # Выполняется после всех add_middleware ДО, первым ПОСЛЕ
...from fastapi.testclient import TestClient
client = TestClient(app)
def test_rate_limit_headers():
response = client.get('/')
assert 'X-RateLimit-Remaining' in response.headers
assert 'X-Process-Time' in response.headers
assert 'X-Content-Type-Options' in response.headers
def test_cors_headers():
response = client.get('/', headers={'Origin': 'https://myapp.com'})
assert 'access-control-allow-origin' in response.headers
def test_gzip():
response = client.get('/large', headers={'Accept-Encoding': 'gzip'})
assert response.headers.get('Content-Encoding') == 'gzip'@app.middleware("http")
async def bad_middleware(request: Request, call_next):
time.sleep(1) # Блокирует все запросы!
return await call_next(request)Решение: Используйте await asyncio.sleep(1).
@app.middleware("http")
async def bad_middleware(request: Request, call_next):
# Забыли call_next!
return Response("Always same response")Решение: Всегда вызывайте response = await call_next(request).
@app.middleware("http")
def custom_middleware(request, call_next):
response = call_next(request)
response.headers['Access-Control-Allow-Origin'] = '*' # Не работает!
return responseРешение: Используйте CORSMiddleware от FastAPI, он обрабатывает preflight запросы.
@app.middleware("http")В следующих темах вы изучите асинхронные паттерны, оптимизацию производительности, безопасность и другие продвинутые возможности FastAPI.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.