Настройка для production, Uvicorn и Gunicorn, мониторинг, логирование, масштабирование, best practices
Настройка для production, Uvicorn и Gunicorn, мониторинг, логирование, масштабирование
Uvicorn — ASGI-сервер на основе uvloop и httptools.
Development:
uvicorn main:app --reload --host 0.0.0.0 --port 8000Production:
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4Параметры:
--reload — автоперезагрузка при изменении кода (только development)--workers — количество рабочих процессов--loop uvloop — использовать uvloop (быстрее)--http httptools — использовать httptools (быстрее)Для production рекомендуется Gunicorn с Uvicorn workers:
gunicorn main:app \
-w 4 \
-k uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 120 \
--keep-alive 5 \
--access-logfile - \
--error-logfile -Параметры:
-w — количество workers (формула: 2 × CPU + 1)-k — класс worker'а--bind — адрес и порт--timeout — таймаут запроса в секундах--keep-alive — время keep-alive соединения--access-logfile — файл логов доступа--error-logfile — файл логов ошибок# gunicorn.conf.py
import multiprocessing
# Server socket
bind = "0.0.0.0:8000"
# Worker processes
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
timeout = 120
keepalive = 5
# Logging
accesslog = "-"
errorlog = "-"
loglevel = "info"
# Process naming
proc_name = "starlite_app"
# Server mechanics
daemon = False
pidfile = None
umask = 0
user = None
group = None
tmp_upload_dir = NoneЗапуск:
gunicorn -c gunicorn.conf.py main:appFROM python:3.11-slim
# Устанавливаем зависимости системы
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
# Создаём пользователя
RUN useradd --create-home --shell /bin/bash app
# Рабочая директория
WORKDIR /home/app
# Устанавливаем Poetry
RUN pip install poetry
# Копируем файлы зависимостей
COPY pyproject.toml poetry.lock ./
# Устанавливаем зависимости
RUN poetry config virtualenvs.create false \
&& poetry install --no-dev --no-interaction
# Копируем приложение
COPY . .
# Переключаемся на пользователя
USER app
# Переменные окружения
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
# Экспортируем порт
EXPOSE 8000
# Команда запуска
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
restart: unless-stopped
db:
image: postgres:15-alpine
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
- POSTGRES_DB=mydb
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
restart: unless-stopped
volumes:
postgres_data:
redis_data:# .env
DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/mydb
REDIS_URL=redis://localhost:6379
SECRET_KEY=your-secret-key-here
DEBUG=false
LOG_LEVEL=info
WORKERS=4import os
from starlite import Starlite
DATABASE_URL = os.getenv("DATABASE_URL")
SECRET_KEY = os.getenv("SECRET_KEY")
DEBUG = os.getenv("DEBUG", "false").lower() == "true"
LOG_LEVEL = os.getenv("LOG_LEVEL", "info")
app = Starlite(
route_handlers=[...],
debug=DEBUG,
)from pydantic import BaseSettings, Field
class Settings(BaseSettings):
database_url: str = Field(env="DATABASE_URL")
redis_url: str = Field(env="REDIS_URL")
secret_key: str = Field(env="SECRET_KEY")
debug: bool = Field(env="DEBUG", default=False)
log_level: str = Field(env="LOG_LEVEL", default="info")
workers: int = Field(env="WORKERS", default=4)
class Config:
env_file = ".env"
settings = Settings()# logging_config.py
import logging
LOGGING_CONFIG = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"default": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
},
"access": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "default",
"level": "INFO",
},
"file": {
"class": "logging.FileHandler",
"formatter": "default",
"filename": "app.log",
"level": "DEBUG",
},
},
"root": {
"handlers": ["console", "file"],
"level": "INFO",
},
"loggers": {
"starlite": {
"handlers": ["console"],
"level": "INFO",
"propagate": False,
},
"uvicorn": {
"handlers": ["console"],
"level": "INFO",
"propagate": False,
},
},
}import logging
from starlite import Starlite
logger = logging.getLogger(__name__)
@get("/users")
def get_users() -> list[User]:
logger.info("Getting all users")
try:
users = db.get_all_users()
logger.debug(f"Found {len(users)} users")
return users
except Exception as e:
logger.exception("Error getting users")
raise
app = Starlite(
route_handlers=[...],
logging_config=LOGGING_CONFIG,
)import json
import logging
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
if record.exc_info:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.getLogger().addHandler(handler)from starlite import get, Response
@get("/health")
def health_check() -> dict:
return {
"status": "healthy",
"version": "1.0.0",
}
@get("/ready")
def readiness_check(db: AsyncSession) -> Response:
try:
# Проверка БД
from sqlalchemy import text
db.execute(text("SELECT 1"))
# Проверка Redis
redis.ping()
return Response(
content={"status": "ready"},
status_code=200,
)
except Exception as e:
return Response(
content={"status": "not ready", "error": str(e)},
status_code=503,
)from prometheus_client import Counter, Histogram, generate_latest
from starlite import get, Response
REQUEST_COUNT = Counter(
"http_requests_total",
"Total HTTP requests",
["method", "endpoint", "status"],
)
REQUEST_DURATION = Histogram(
"http_request_duration_seconds",
"HTTP request duration",
)
@get("/metrics")
def metrics() -> Response:
return Response(
content=generate_latest(),
media_type="text/plain",
)# docker-compose.prod.yml
version: '3.8'
services:
app:
image: myapp:latest
deploy:
replicas: 4
resources:
limits:
cpus: '0.5'
memory: 512M
environment:
- DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
depends_on:
- db
- redis
db:
image: postgres:15-alpine
# Настройка для production
command: >
postgres
-c max_connections=200
-c shared_buffers=256MB
-c effective_cache_size=768MB
redis:
image: redis:7-alpine
command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lrufrom starlite import get
import redis.asyncio as redis
redis_client = redis.Redis(host="localhost", port=6379)
@get("/items/{item_id:int}")
async def get_item(
item_id: int,
db: AsyncSession,
) -> Item:
# Проверка кэша
cached = await redis_client.get(f"item:{item_id}")
if cached:
return Item.parse_raw(cached)
# Запрос к БД
result = await db.execute(
select(Item).where(Item.id == item_id)
)
item = result.scalar_one()
# Сохранение в кэш
await redis_client.setex(
f"item:{item_id}",
300, # 5 минут
item.json(),
)
return itemfrom starlite import Request, Response
from starlite.exceptions import TooManyRequests
import redis.asyncio as redis
redis_client = redis.Redis()
async def rate_limit_middleware(request: Request, call_next):
client_ip = request.client.host
key = f"rate_limit:{client_ip}"
# Счётчик запросов
count = await redis_client.incr(key)
if count == 1:
await redis_client.expire(key, 60) # 1 минута
if count > 100: # 100 запросов в минуту
raise TooManyRequests(
detail="Rate limit exceeded",
headers={"Retry-After": "60"},
)
response = await call_next(request)
response.headers["X-RateLimit-Remaining"] = str(100 - count)
return response# Запуск с SSL
uvicorn main:app \
--host 0.0.0.0 \
--port 443 \
--ssl-certfile=/path/to/cert.pem \
--ssl-keyfile=/path/to/key.pemfrom starlite import Response
async def security_headers_middleware(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"
response.headers["Strict-Transport-Security"] = "max-age=31536000"
response.headers["Content-Security-Policy"] = "default-src 'self'"
return responsefrom starlite.middleware import CORSConfig
cors_config = CORSConfig(
allow_origins=["https://example.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
max_age=600,
)
app = Starlite(
route_handlers=[...],
cors_config=cors_config,
)# .github/workflows/deploy.yml
name: Deploy
on:
push:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: '3.11'
- run: pip install poetry
- run: poetry install
- run: poetry run pytest
deploy:
needs: test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Build and push Docker image
run: |
docker build -t myapp:latest .
docker push myapp:latest
- name: Deploy to server
run: |
ssh user@server "docker pull myapp:latest && docker-compose up -d"# ✅ Хорошо
DATABASE_URL = os.getenv("DATABASE_URL")
# ❌ Плохо
DATABASE_URL = "postgresql://localhost/mydb"# ✅ Хорошо
DEBUG = os.getenv("DEBUG", "false").lower() == "true"
# ❌ Плохо
DEBUG = True# ✅ Хорошо
logger.info("User created", extra={"user_id": user.id})
# ❌ Плохо
print("User created")# ✅ Хорошо
@get("/health")
def health_check() -> dict:
return {"status": "healthy"}
# ❌ Плохо — нет health check# ✅ Хорошо
engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=40,
)
# ❌ Плохо — нет пулинга
engine = create_async_engine(DATABASE_URL)Production настройка Starlite включает:
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.