Sanic – быстрый асинхронный веб-фреймворк

Введение в Sanic

Sanic — это современный асинхронный веб-фреймворк для Python, разработанный для создания высокопроизводительных веб-приложений. Основанный на asyncio, он позволяет эффективно обрабатывать тысячи одновременных соединений благодаря поддержке async/await синтаксиса. Sanic идеально подходит для создания REST API, микросервисов, real-time приложений и бэкендов с высокой нагрузкой.

Ключевые особенности Sanic

Производительность и архитектура

Sanic построен с акцентом на скорость выполнения. Фреймворк использует asyncio для асинхронной обработки запросов, что позволяет достичь производительности, сравнимой с Node.js и Go. Встроенная поддержка async/await делает код более читаемым и эффективным.

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

• Полная поддержка async/await из коробки
• Маршрутизация с параметрами и middleware
• Встроенная поддержка WebSocket
• Система шаблонов и работа со статическими файлами
• Поддержка CORS, cookie и сессий
• Простая и понятная структура проекта
• Отличная масштабируемость для API-серверов

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

Базовая установка

pip install sanic

Дополнительные зависимости

pip install sanic[ext]  # Расширения
pip install jinja2 aiofiles python-dotenv

Создание первого приложения

from sanic import Sanic
from sanic.response import text

app = Sanic("MyApp")

@app.get("/")
async def index(request):
    return text("Привет от Sanic")

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

Маршрутизация и обработка запросов

Базовые маршруты

@app.get("/")
async def get_handler(request):
    return text("GET запрос")

@app.post("/")
async def post_handler(request):
    return text("POST запрос")

@app.put("/")
async def put_handler(request):
    return text("PUT запрос")

@app.delete("/")
async def delete_handler(request):
    return text("DELETE запрос")

Асинхронные маршруты

import asyncio

@app.get("/async")
async def async_route(request):
    await asyncio.sleep(1)  # Имитация асинхронной операции
    return text("Асинхронный ответ")

Работа с параметрами и данными

Параметры маршрута

@app.get("/user/<user_id:int>")
async def get_user(request, user_id):
    return text(f"ID пользователя: {user_id}")

@app.get("/profile/<username:str>")
async def get_profile(request, username):
    return text(f"Профиль: {username}")

Query-параметры

@app.get("/search")
async def search(request):
    query = request.args.get("q", "ничего")
    limit = request.args.get("limit", "10")
    return text(f"Поиск: {query}, лимит: {limit}")

Обработка JSON и форм

from sanic.response import json

@app.post("/api/data")
async def receive_data(request):
    data = request.json
    return json({"received": data})

@app.post("/form")
async def handle_form(request):
    username = request.form.get("username")
    email = request.form.get("email")
    return json({"username": username, "email": email})

Типы ответов

Различные форматы ответов

from sanic.response import text, json, html, redirect, file

@app.get("/text")
async def text_response(request):
    return text("Простой текст")

@app.get("/json")
async def json_response(request):
    return json({"message": "JSON ответ"})

@app.get("/html")
async def html_response(request):
    return html("<h1>HTML страница</h1>")

@app.get("/redirect")
async def redirect_response(request):
    return redirect("/new-page")

@app.get("/file")
async def file_response(request):
    return await file("path/to/file.pdf")

Обработка ошибок

Встроенные исключения

from sanic.exceptions import NotFound, SanicException, Unauthorized

@app.exception(NotFound)
async def handle_404(request, exception):
    return json({"error": "Страница не найдена"}, status=404)

@app.exception(Exception)
async def handle_generic_error(request, exception):
    return json({"error": "Внутренняя ошибка сервера"}, status=500)

Кастомные исключения

class CustomException(SanicException):
    status_code = 400
    message = "Кастомная ошибка"

@app.exception(CustomException)
async def handle_custom_error(request, exception):
    return json({"error": exception.message}, status=exception.status_code)

Middleware и жизненный цикл

Request и Response Middleware

@app.middleware("request")
async def before_request(request):
    request.ctx.start_time = time.time()
    print(f"Запрос к {request.path}")

@app.middleware("response")
async def after_response(request, response):
    response.headers["X-Framework"] = "Sanic"
    response.headers["X-Process-Time"] = str(time.time() - request.ctx.start_time)

Lifecycle хуки

@app.before_server_start
async def setup_db(app, loop):
    print("Подключение к базе данных")

@app.after_server_stop
async def close_db(app, loop):
    print("Закрытие соединения с базой данных")

Работа с cookie и сессиями

Cookie

@app.get("/set-cookie")
async def set_cookie(request):
    response = text("Cookie установлен")
    response.cookies["session_id"] = "123456"
    response.cookies["session_id"]["httponly"] = True
    response.cookies["session_id"]["secure"] = True
    return response

@app.get("/get-cookie")
async def get_cookie(request):
    session_id = request.cookies.get("session_id", "не установлен")
    return text(f"Session ID: {session_id}")

Работа с заголовками

@app.get("/headers")
async def handle_headers(request):
    user_agent = request.headers.get("User-Agent")
    auth_header = request.headers.get("Authorization")
    
    response = json({"user_agent": user_agent})
    response.headers["Custom-Header"] = "Sanic-Value"
    return response

Загрузка файлов

Обработка файлов

import os

@app.post("/upload")
async def upload_file(request):
    file = request.files.get("file")
    if file:
        upload_dir = "./uploads"
        os.makedirs(upload_dir, exist_ok=True)
        
        file_path = os.path.join(upload_dir, file.name)
        with open(file_path, "wb") as f:
            f.write(file.body)
        
        return json({"message": "Файл загружен", "filename": file.name})
    return json({"error": "Файл не найден"}, status=400)

Шаблоны и статические файлы

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

from sanic_ext import Extend
from jinja2 import Environment, FileSystemLoader

app = Sanic("TemplateApp")
Extend(app)

@app.get("/template")
async def render_template(request):
    return await app.ext.render("index.html", context={"name": "Sanic"})

Статические файлы

app.static("/static", "./static")
app.static("/media", "./media", name="media")

WebSocket поддержка

Базовый WebSocket

@app.websocket("/ws")
async def websocket_handler(request, ws):
    while True:
        try:
            message = await ws.recv()
            await ws.send(f"Эхо: {message}")
        except:
            break

WebSocket с аутентификацией

@app.websocket("/ws/auth")
async def auth_websocket(request, ws):
    token = request.headers.get("Authorization")
    if not token:
        await ws.close(code=1008, reason="Unauthorized")
        return
    
    while True:
        message = await ws.recv()
        await ws.send(f"Authenticated echo: {message}")

Интеграция с базами данных

Использование с SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///./database.db")
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession)

@app.middleware("request")
async def inject_session(request):
    request.ctx.session = SessionLocal()

@app.middleware("response")
async def close_session(request, response):
    await request.ctx.session.close()

Работа с async/await ORM

@app.get("/users")
async def get_users(request):
    async with request.ctx.session as session:
        # Пример работы с ORM
        users = await session.execute("SELECT * FROM users")
        return json([dict(user) for user in users])

Структура проекта

Рекомендуемая организация

project/
├── app.py              # Основное приложение
├── config.py           # Конфигурация
├── requirements.txt    # Зависимости
├── routes/            # Маршруты
│   ├── __init__.py
│   ├── api.py
│   └── auth.py
├── models/            # Модели данных
│   ├── __init__.py
│   └── user.py
├── middleware/        # Middleware
│   ├── __init__.py
│   └── auth.py
├── templates/         # Шаблоны
│   └── index.html
├── static/           # Статические файлы
│   ├── css/
│   └── js/
└── tests/            # Тесты
    ├── __init__.py
    └── test_app.py

Blueprints для модульности

Создание Blueprint

from sanic import Blueprint

# routes/users.py
users_bp = Blueprint("users", url_prefix="/users")

@users_bp.get("/")
async def get_users(request):
    return json({"users": []})

@users_bp.get("/<user_id:int>")
async def get_user(request, user_id):
    return json({"user_id": user_id})

# app.py
from routes.users import users_bp
app.blueprint(users_bp)

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

Написание тестов

import pytest
from app import app

@pytest.mark.asyncio
async def test_index():
    request, response = await app.asgi_client.get("/")
    assert response.status == 200
    assert "Привет от Sanic" in response.text

@pytest.mark.asyncio
async def test_json_endpoint():
    request, response = await app.asgi_client.get("/api/data")
    assert response.status == 200
    assert response.json["status"] == "ok"

Тестирование с фикстурами

@pytest.fixture
def test_client():
    return app.asgi_client

@pytest.mark.asyncio
async def test_with_fixture(test_client):
    request, response = await test_client.post("/api/users", json={"name": "Test"})
    assert response.status == 201

Развертывание и продакшн

Настройки для продакшн

# config.py
class ProductionConfig:
    DEBUG = False
    ACCESS_LOG = False
    WORKERS = 4
    HOST = "0.0.0.0"
    PORT = 8000

# app.py
from config import ProductionConfig

if __name__ == "__main__":
    app.run(
        host=ProductionConfig.HOST,
        port=ProductionConfig.PORT,
        workers=ProductionConfig.WORKERS,
        debug=ProductionConfig.DEBUG,
        access_log=ProductionConfig.ACCESS_LOG
    )

Запуск через командную строку

# Базовый запуск
sanic app.app --host=0.0.0.0 --port=8000

# Продакшн с воркерами
sanic app.app --host=0.0.0.0 --port=8000 --workers=4

# С дополнительными опциями
sanic app.app --workers=4 --access-log --debug

Использование с Gunicorn

pip install gunicorn
gunicorn app:app --bind 0.0.0.0:8000 --worker-class sanic.worker.GunicornWorker

Полная таблица методов и функций Sanic

Расширенные возможности

Кастомные декораторы

from functools import wraps

def require_auth(f):
    @wraps(f)
    async def decorated_function(request, *args, **kwargs):
        auth = request.headers.get("Authorization")
        if not auth:
            return json({"error": "Unauthorized"}, status=401)
        return await f(request, *args, **kwargs)
    return decorated_function

@app.get("/protected")
@require_auth
async def protected_route(request):
    return json({"message": "Protected content"})

Валидация данных

from cerberus import Validator

@app.post("/api/users")
async def create_user(request):
    schema = {
        'name': {'type': 'string', 'required': True},
        'email': {'type': 'string', 'required': True, 'regex': r'^[^@]+@[^@]+\.[^@]+$'},
        'age': {'type': 'integer', 'min': 0, 'max': 150}
    }
    
    v = Validator(schema)
    if not v.validate(request.json):
        return json({"errors": v.errors}, status=400)
    
    # Обработка валидных данных
    return json({"message": "User created"}, status=201)

Практические примеры применения

RESTful API

users_data = []

@app.get("/api/users")
async def get_users(request):
    return json(users_data)

@app.post("/api/users")
async def create_user(request):
    user = request.json
    user['id'] = len(users_data) + 1
    users_data.append(user)
    return json(user, status=201)

@app.get("/api/users/<user_id:int>")
async def get_user(request, user_id):
    user = next((u for u in users_data if u['id'] == user_id), None)
    if not user:
        return json({"error": "User not found"}, status=404)
    return json(user)

WebSocket чат

connected_clients = set()

@app.websocket("/chat")
async def chat_handler(request, ws):
    connected_clients.add(ws)
    try:
        async for message in ws:
            # Отправляем сообщение всем подключенным клиентам
            for client in connected_clients.copy():
                try:
                    await client.send(message)
                except:
                    connected_clients.remove(client)
    finally:
        connected_clients.discard(ws)

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

Настройки для высокой нагрузки

# Оптимизированная конфигурация
app.config.update({
    "KEEP_ALIVE_TIMEOUT": 30,
    "REQUEST_TIMEOUT": 60,
    "RESPONSE_TIMEOUT": 60,
    "REQUEST_MAX_SIZE": 100000000,  # 100MB
    "ACCESS_LOG": False,
    "AUTO_RELOAD": False,
})

Использование connection pooling

import aioredis
from aiohttp import ClientSession

@app.before_server_start
async def setup_connections(app, loop):
    # Redis connection pool
    app.ctx.redis = await aioredis.create_redis_pool(
        'redis://localhost',
        maxsize=10
    )
    
    # HTTP client session
    app.ctx.http_session = ClientSession()

@app.after_server_stop
async def cleanup_connections(app, loop):
    app.ctx.redis.close()
    await app.ctx.redis.wait_closed()
    await app.ctx.http_session.close()

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

CORS настройки

from sanic_cors import CORS

CORS(app, resources={
    r"/api/*": {
        "origins": ["http://localhost:3000"],
        "methods": ["GET", "POST", "PUT", "DELETE"],
        "allow_headers": ["Content-Type", "Authorization"]
    }
})

Ограничение частоты запросов

from collections import defaultdict
import time

request_counts = defaultdict(list)

@app.middleware("request")
async def rate_limit(request):
    client_ip = request.ip
    now = time.time()
    
    # Очистка старых запросов
    request_counts[client_ip] = [
        req_time for req_time in request_counts[client_ip]
        if now - req_time < 60  # За последнюю минуту
    ]
    
    # Проверка лимита
    if len(request_counts[client_ip]) >= 100:  # 100 запросов в минуту
        return json({"error": "Rate limit exceeded"}, status=429)
    
    request_counts[client_ip].append(now)

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

Что такое Sanic и для чего он используется?

Sanic — это высокопроизводительный асинхронный веб-фреймворк для Python, использующий async/await синтаксис. Он предназначен для создания быстрых API, веб-сервисов и микросервисов с возможностью обработки большого количества одновременных соединений.

Чем Sanic отличается от Flask и Django?

В отличие от Flask и Django, Sanic полностью асинхронный, что обеспечивает значительно более высокую производительность при обработке множественных запросов. Он также имеет встроенную поддержку WebSocket и более современный подход к разработке API.

Можно ли использовать Sanic в продакшн среде?

Да, Sanic подходит для продакшн использования. Многие компании успешно используют его для высоконагруженных приложений. Для продакшн рекомендуется использовать несколько воркеров и настроить соответствующие параметры производительности.

Поддерживает ли Sanic работу с базами данных?

Sanic отлично работает с асинхронными ORM и драйверами баз данных, такими как SQLAlchemy (с asyncio), asyncpg для PostgreSQL, aiomysql для MySQL и другими async-совместимыми библиотеками.

Как обеспечить безопасность в Sanic приложениях?

Sanic поддерживает стандартные методы безопасности: HTTPS, CORS, аутентификацию через JWT, валидацию входящих данных, ограничение частоты запросов и использование middleware для проверки прав доступа.

Какие инструменты тестирования доступны для Sanic?

Sanic предоставляет встроенный тестовый клиент, поддерживает pytest для асинхронных тестов и имеет отличную совместимость с популярными библиотеками для тестирования Python приложений.