Uvicorn – ASGI-сервер для FastAPI и Django

Uvicorn: Полное руководство по высокопроизводительному ASGI-серверу

Что такое Uvicorn

Uvicorn — это современный, легковесный и высокопроизводительный ASGI-сервер, написанный на Python. Он специально разработан для запуска асинхронных веб-приложений и является одним из самых быстрых серверов для Python-приложений. Uvicorn построен на базе библиотек uvloop и httptools, что обеспечивает исключительную производительность при обработке большого количества одновременных запросов.

Сервер поддерживает современные веб-протоколы и технологии, включая HTTP/1.1, WebSocket, и может работать с любыми ASGI-совместимыми фреймворками, такими как FastAPI, Starlette, Django Channels и другими.

Основные преимущества и особенности

Архитектурные преимущества

Uvicorn представляет собой значительный шаг вперед в области веб-серверов Python благодаря следующим особенностям:

Асинхронность на уровне ядра — использует событийно-ориентированную архитектуру, позволяющую обрабатывать тысячи одновременных соединений с минимальными затратами ресурсов.

Высокая производительность — благодаря интеграции с uvloop (быстрая реализация asyncio) и httptools (быстрый HTTP-парсер), Uvicorn демонстрирует производительность, сравнимую с серверами на Go и Node.js.

Поддержка современных протоколов — нативная поддержка WebSocket, HTTP/1.1, а также возможность работы с HTTP/2 через дополнительные прокси.

Минимальные системные требования — оптимизирован для работы как на локальных машинах разработчиков, так и на продакшн-серверах.

Технические особенности

Uvicorn реализует спецификацию ASGI 3.0, что делает его совместимым с широким спектром современных Python-фреймворков. Сервер поддерживает автоматическое обнаружение изменений в коде во время разработки, многопроцессорный режим для продакшена и гибкую систему конфигурации.

Сравнение ASGI и WSGI

Понимание различий между ASGI и WSGI критически важно для выбора подходящего сервера:

ASGI является эволюционным развитием WSGI, предназначенным для современных веб-приложений, требующих высокой производительности и поддержки real-time функций.

Установка и первоначальная настройка

Установка через pip

pip install uvicorn

Установка с дополнительными зависимостями

# Для максимальной производительности
pip install uvicorn[standard]

# Включает uvloop, httptools, websockets, watchgod, python-dotenv

Минимальный пример приложения

Создайте файл main.py:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Привет от Uvicorn", "status": "working"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Запуск сервера

# Базовый запуск
uvicorn main:app

# Запуск с автоперезагрузкой для разработки
uvicorn main:app --reload

# Запуск на всех интерфейсах
uvicorn main:app --host 0.0.0.0 --port 8000

Детальный обзор параметров командной строки

Основные параметры запуска

uvicorn main:app \
  --host 0.0.0.0 \
  --port 8000 \
  --reload \
  --workers 4 \
  --log-level info

Полная таблица параметров

Программный запуск Uvicorn

Базовый программный запуск

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Программный запуск Uvicorn"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Расширенная конфигурация

import uvicorn
from fastapi import FastAPI

app = FastAPI()

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info",
        access_log=True,
        workers=1,  # workers > 1 несовместимо с reload=True
        loop="asyncio",
        http="auto",
        ws="auto",
        lifespan="auto",
        ssl_keyfile=None,
        ssl_certfile=None,
        proxy_headers=False,
        forwarded_allow_ips=None,
        root_path="",
        timeout_keep_alive=5,
        timeout_notify=30,
        limit_concurrency=None,
        limit_max_requests=None,
        backlog=2048,
        reload_dirs=None,
        reload_delay=0.25,
        reload_excludes=None,
        reload_includes=None
    )

Интеграция с различными фреймворками

FastAPI

FastAPI и Uvicorn созданы для совместной работы:

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI(title="FastAPI + Uvicorn", version="1.0.0")

@app.get("/")
async def root():
    return {"framework": "FastAPI", "server": "Uvicorn"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id, "server": "Uvicorn"}

# Запуск: uvicorn main:app --reload

Starlette

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({"message": "Starlette + Uvicorn"})

routes = [
    Route("/", homepage),
]

app = Starlette(routes=routes)

# Запуск: uvicorn main:app --reload

Django ASGI

Для Django 3.0+:

# asgi.py
import os
from django.core.asgi import get_asgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = get_asgi_application()

# Запуск Django через Uvicorn
uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000

Quart

from quart import Quart, jsonify

app = Quart(__name__)

@app.route("/")
async def hello():
    return jsonify({"message": "Quart + Uvicorn"})

# Запуск: uvicorn main:app --reload

Работа с WebSocket

Базовый WebSocket с FastAPI

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
<head>
    <title>WebSocket Test</title>
</head>
<body>
    <h1>WebSocket Test</h1>
    <input type="text" id="messageInput" placeholder="Введите сообщение">
    <button onclick="sendMessage()">Отправить</button>
    <ul id="messages"></ul>
    <script>
        const ws = new WebSocket("ws://localhost:8000/ws");
        
        ws.onmessage = function(event) {
            const messages = document.getElementById('messages');
            const message = document.createElement('li');
            message.textContent = event.data;
            messages.appendChild(message);
        };
        
        function sendMessage() {
            const input = document.getElementById('messageInput');
            ws.send(input.value);
            input.value = '';
        }
    </script>
</body>
</html>
"""

@app.get("/")
async def get():
    return HTMLResponse(html)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Клиент отключился")

Продвинутый WebSocket с управлением соединениями

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            await manager.send_personal_message(f"Client {client_id}: {message['text']}", websocket)
            await manager.broadcast(f"Client {client_id} says: {message['text']}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} left the chat")

Конфигурация для продакшена

Системный сервис (systemd)

Создайте файл /etc/systemd/system/myapp.service:

[Unit]
Description=MyApp Uvicorn Service
After=network.target

[Service]
Type=exec
User=www-data
Group=www-data
WorkingDirectory=/var/www/myapp
ExecStart=/var/www/myapp/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
Restart=always
RestartSec=10
StandardOutput=syslog
StandardError=syslog
SyslogIdentifier=myapp

[Install]
WantedBy=multi-user.target

Активация:

sudo systemctl daemon-reload
sudo systemctl enable myapp
sudo systemctl start myapp

Supervisor

Конфигурация в /etc/supervisor/conf.d/myapp.conf:

[program:myapp]
command=/var/www/myapp/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
directory=/var/www/myapp
user=www-data
autostart=true
autorestart=true
stdout_logfile=/var/log/myapp/uvicorn.log
stderr_logfile=/var/log/myapp/uvicorn.error.log

Docker контейнер

FROM python:3.11-slim

WORKDIR /app

# Копируем requirements и устанавливаем зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копируем код приложения
COPY . .

# Создаем пользователя для безопасности
RUN useradd --create-home --shell /bin/bash app && chown -R app:app /app
USER app

# Настраиваем переменные окружения
ENV PYTHONPATH=/app
ENV PYTHONUNBUFFERED=1

# Открываем порт
EXPOSE 8000

# Команда по умолчанию
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Docker Compose

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - PYTHONUNBUFFERED=1
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      - db
    volumes:
      - ./app:/app
    restart: unless-stopped

  db:
    image: postgres:13
    environment:
      - POSTGRES_DB=mydb
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - web
    restart: unless-stopped

volumes:
  postgres_data:

Настройка обратного прокси

Nginx конфигурация

upstream uvicorn {
    server 127.0.0.1:8000;
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    server 127.0.0.1:8003;
}

server {
    listen 80;
    server_name example.com;
    
    # Перенаправление на HTTPS
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name example.com;
    
    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;
    
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;
    ssl_prefer_server_ciphers on;
    
    client_max_body_size 64M;
    
    location / {
        proxy_pass http://uvicorn;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_redirect off;
        proxy_buffering off;
    }
    
    location /ws {
        proxy_pass http://uvicorn;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_cache_bypass $http_upgrade;
    }
}

Traefik конфигурация

# docker-compose.yml
version: '3.8'

services:
  traefik:
    image: traefik:v2.10
    command:
      - "--providers.docker=true"
      - "--providers.docker.exposedbydefault=false"
      - "--entrypoints.web.address=:80"
      - "--entrypoints.websecure.address=:443"
      - "--certificatesresolvers.myresolver.acme.tlschallenge=true"
      - "--certificatesresolvers.myresolver.acme.email=admin@example.com"
      - "--certificatesresolvers.myresolver.acme.storage=/letsencrypt/acme.json"
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock:ro
      - ./letsencrypt:/letsencrypt

  web:
    build: .
    labels:
      - "traefik.enable=true"
      - "traefik.http.routers.web.rule=Host(`example.com`)"
      - "traefik.http.routers.web.entrypoints=websecure"
      - "traefik.http.routers.web.tls.certresolver=myresolver"
      - "traefik.http.services.web.loadbalancer.server.port=8000"
    restart: unless-stopped

Мониторинг и логирование

Структурированное логирование

import logging
import json
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

app = FastAPI()

class StructuredLogger:
    def __init__(self):
        self.logger = logging.getLogger("structured")
        
    def log_request(self, request: Request, response_time: float, status_code: int):
        log_data = {
            "timestamp": time.time(),
            "method": request.method,
            "url": str(request.url),
            "user_agent": request.headers.get("user-agent"),
            "ip": request.client.host,
            "response_time": response_time,
            "status_code": status_code
        }
        self.logger.info(json.dumps(log_data))

structured_logger = StructuredLogger()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    structured_logger.log_request(request, process_time, response.status_code)
    return response

@app.get("/")
async def root():
    return {"message": "Логирование работает"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

Интеграция с Prometheus

from prometheus_client import Counter, Histogram, generate_latest
from fastapi import FastAPI, Request, Response
from fastapi.responses import PlainTextResponse
import time

REQUEST_COUNT = Counter('requests_total', 'Total requests', ['method', 'endpoint'])
REQUEST_LATENCY = Histogram('request_duration_seconds', 'Request latency')

app = FastAPI()

@app.middleware("http")
async def prometheus_middleware(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    
    REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path).inc()
    REQUEST_LATENCY.observe(process_time)
    
    return response

@app.get("/metrics")
async def metrics():
    return PlainTextResponse(generate_latest())

@app.get("/")
async def root():
    return {"message": "Метрики доступны на /metrics"}

Тестирование приложений

Тестирование с pytest

import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Привет от Uvicorn", "status": "working"}

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "healthy"}

@pytest.mark.asyncio
async def test_websocket():
    with client.websocket_connect("/ws") as websocket:
        websocket.send_text("test message")
        data = websocket.receive_text()
        assert data == "Echo: test message"

Нагрузочное тестирование

import asyncio
import aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

async def load_test():
    url = "http://localhost:8000/"
    tasks = []
    
    async with aiohttp.ClientSession() as session:
        start_time = time.time()
        
        for i in range(1000):
            task = asyncio.create_task(fetch(session, url))
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        end_time = time.time()
        
        print(f"Выполнено {len(results)} запросов за {end_time - start_time:.2f} секунд")
        print(f"RPS: {len(results) / (end_time - start_time):.2f}")

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

Оптимизация производительности

Настройка рабочих процессов

import multiprocessing
import uvicorn
from fastapi import FastAPI

app = FastAPI()

def get_optimal_workers():
    """Определяет оптимальное количество воркеров"""
    cpu_count = multiprocessing.cpu_count()
    return (cpu_count * 2) + 1

@app.get("/")
async def root():
    return {"message": "Оптимизированный сервер"}

if __name__ == "__main__":
    workers = get_optimal_workers()
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        workers=workers,
        loop="uvloop",
        http="httptools",
        access_log=False,  # Отключаем для производительности
        timeout_keep_alive=30
    )

Оптимизация для конкретных задач

import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

# Для высокой нагрузки
PRODUCTION_CONFIG = {
    "host": "0.0.0.0",
    "port": 8000,
    "workers": 4,
    "loop": "uvloop",
    "http": "httptools",
    "access_log": False,
    "timeout_keep_alive": 30,
    "limit_concurrency": 1000,
    "backlog": 2048
}

# Для WebSocket приложений
WEBSOCKET_CONFIG = {
    "host": "0.0.0.0",
    "port": 8000,
    "workers": 1,  # WebSocket требует один процесс
    "loop": "uvloop",
    "ws": "websockets",
    "timeout_keep_alive": 60
}

@app.get("/")
async def root():
    return JSONResponse({"message": "Высокопроизводительный сервер"})

if __name__ == "__main__":
    uvicorn.run("main:app", **PRODUCTION_CONFIG)

Методы и функции Uvicorn

Основные методы и классы

Подробная таблица параметров uvicorn.run()

Программные интерфейсы

import uvicorn
from fastapi import FastAPI

app = FastAPI()

# Метод 1: Простой запуск
uvicorn.run(app)

# Метод 2: Конфигурация через Config
config = uvicorn.Config(
    app,
    host="0.0.0.0",
    port=8000,
    log_level="info",
    access_log=True
)
server = uvicorn.Server(config)

# Метод 3: Асинхронный запуск
import asyncio

async def main():
    config = uvicorn.Config(app, host="0.0.0.0", port=8000)
    server = uvicorn.Server(config)
    await server.serve()

# Метод 4: Управление жизненным циклом
async def startup():
    config = uvicorn.Config(app, host="0.0.0.0", port=8000)
    server = uvicorn.Server(config)
    
    # Запуск в фоновом режиме
    await server.startup()
    
    # Остановка
    await server.shutdown()

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

Сравнение с другими серверами

Производительность

Функциональные возможности

Безопасность

Конфигурация SSL/TLS

import uvicorn
from fastapi import FastAPI

app = FastAPI()

# SSL конфигурация
SSL_CONFIG = {
    "ssl_keyfile": "/path/to/private.key",
    "ssl_certfile": "/path/to/certificate.crt",
    "ssl_ca_certs": "/path/to/ca-bundle.crt",
    "ssl_version": ssl.PROTOCOL_TLSv1_2,
    "ssl_cert_reqs": ssl.CERT_REQUIRED,
    "ssl_ciphers": "HIGH:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!SRP:!CAMELLIA"
}

@app.get("/")
async def root():
    return {"message": "Безопасное соединение"}

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=443,
        **SSL_CONFIG
    )

Защита от атак

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

app = FastAPI()

# Ограничение скорости запросов
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

# Проверка доверенных хостов
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

@app.get("/")
@limiter.limit("10/minute")
async def root(request: Request):
    return {"message": "Защищенный эндпоинт"}

Часто задаваемые вопросы

Что такое Uvicorn и зачем он нужен?

Uvicorn — это высокопроизводительный ASGI-сервер для Python, предназначенный для запуска асинхронных веб-приложений. Он необходим для запуска современных фреймворков, таких как FastAPI, Starlette, и Django с поддержкой асинхронности. Uvicorn обеспечивает значительно более высокую производительность по сравнению с традиционными WSGI-серверами.

Чем Uvicorn отличается от Gunicorn?

Основные отличия заключаются в архитектуре и возможностях. Uvicorn поддерживает асинхронное программирование, WebSocket, и имеет более высокую производительность благодаря событийно-ориентированной архитектуре. Gunicorn является синхронным WSGI-сервером, но может использовать Uvicorn в качестве воркера для поддержки ASGI.

Можно ли использовать Uvicorn с Django?

Да, начиная с Django 3.0, фреймворк поддерживает ASGI. Вы можете запустить Django через Uvicorn, используя ASGI-приложение Django. Это позволяет использовать асинхронные view-функции и WebSocket в Django.

Как настроить Uvicorn для продакшена?

Для продакшена рекомендуется: использовать несколько воркеров, настроить обратный прокси (Nginx), отключить режим отладки, настроить SSL/TLS, использовать systemd или supervisor для управления процессами, и настроить мониторинг и логирование.

Поддерживает ли Uvicorn HTTP/2?

Uvicorn не поддерживает HTTP/2 напрямую, но может работать за обратным прокси (например, Nginx) с поддержкой HTTP/2. Для нативной поддержки HTTP/2 можно использовать Hypercorn в качестве альтернативы.

Как работает автоперезагрузка в Uvicorn?

Автоперезагрузка активируется флагом --reload и отслеживает изменения в файлах Python. При обнаружении изменений сервер автоматически перезапускается. Эта функция предназначена только для разработки и не должна использоваться в продакшене.

Сколько воркеров нужно для оптимальной производительности?

Рекомендуется использовать формулу: (количество CPU ядер × 2) + 1. Однако оптимальное количество зависит от характера приложения. Для I/O-интенсивных задач может быть эффективно больше воркеров, для CPU-интенсивных — меньше.

Как обрабатывать статические файлы в Uvicorn?

Uvicorn не предназначен для обслуживания статических файлов. Рекомендуется использовать CDN или обратный прокси (Nginx) для статики. В FastAPI можно использовать StaticFiles для разработки, но не для продакшена.

Заключение

Uvicorn представляет собой современное решение для запуска высокопроизводительных веб-приложений на Python. Его асинхронная архитектура, поддержка современных протоколов и отличная интеграция с популярными фреймворками делают его идеальным выбором для разработки современных веб-сервисов.

Сервер обеспечивает отличную производительность, простоту настройки и богатые возможности конфигурации, что делает его подходящим как для разработки, так и для продакшена. Поддержка WebSocket, интеграция с системами мониторинга и возможность масштабирования через множественные воркеры позволяют создавать надежные и производительные веб-приложения.

Выбор Uvicorn в качестве ASGI-сервера обеспечивает долгосрочную совместимость с развивающимися стандартами веб-разработки и гарантирует высокую производительность ваших Python-приложений.