LitestarCatsCV. Тренируемся на кошках. Расширяем возможности и готовимся к продакшену. Часть 3
Котики выходят на новый уровень! 🐾
Привет, котики и котолюбы! В первой части нашего кошачьего приключения мы выбрали инструменты (Litestar вместо FastAPI, Granian вместо Gunicorn, KeyDB вместо Redis), настроили uv и заложили фундамент проекта. Во второй части мы построили полноценное CRUD API для резюме котиков (или людей, если вам так ближе), подружили его с PostgreSQL через SQLAlchemy, настроили миграции с Alembic и написали тесты с Pytest. У нас уже есть стены и фундамент, но пора ставить крышу и готовиться к продакшену! 🏠
Сегодня мы сделаем наш API ещё круче: вынесем конфиги в отдельный модуль с помощью msgspec, добавим аутентификацию через встроенный JWT в Litestar, ускорим API с KeyDB, проверим покрытие тестами с coverage, упакуем всё в Docker и нарисуем резюме котиков с помощью Jinja. К концу статьи наш кошачий проект будет готов к реальной жизни — поехали! 🚀
Если раньше мы просто тренировались на кошках, то теперь выпускаем их в большой мир с пропусками, кешем и стильными резюме. Полный код, как всегда, на GitHub — ссылка в конце! 🐱
Вынос конфигов — приводим все настройки в порядок 🗂️
Зачем это нужно?
Наш проект растёт, и захардкоженные строки вроде DATABASE_URL начинают мяукать от неудобства. Давай вынесем конфиги в отдельный модуль src/configs/app_config.py и будем хранить значения в settings-example.yaml. Мы уже используем msgspec для моделей, так что применим его и для конфигов — это сделает код чище и быстрее, а котики любят порядок! 😺
Что будем делать?
Создадим классы конфигов с помощью msgspec.Struct.
Настроим парсинг settings-example.yaml в эти классы.
Обновим src/app.py, чтобы использовать новые конфиги.
Детали реализации
Установка зависимостей
Нам понадобится pyyaml для парсинга YAML. msgspec у нас уже есть, так как он используется в проекте:
uv add pyyaml
Создаём классы конфигов
Создаём файл src/configs/app_config.py. Используем msgspec.Struct для определения структуры конфигов:
from msgspec import Struct
import argparse
import yaml
from pathlib import Path
class DatabaseConfig(Struct):
user: str = "postgres"
password: str = "postgres"
host: str = "127.0.0.1"
port: str = "5432"
database_name: str = "test_db"
url: str = None
echo: bool = False
def get_connection_url(self) -> str:
if self.url:
return self.url
return f"postgresql+asyncpg://{self.user}:{self.password}@{self.host}:{self.port}/{self.database_name}"
class KeyDBConfig(Struct):
host: str = "127.0.0.1"
port: str = "6379"
db: str = "0"
url: str = None
def get_connection_url(self) -> str:
if self.url:
return self.url
return f"keydb://{self.host}:{self.port}/{self.db}"
class JWTConfig(Struct):
secret: str
token_secret: str
class AppConfig(Struct):
database: DatabaseConfig
keydb: KeyDBConfig
jwt: JWTConfig
def load_config(file_path: str) -> AppConfig:
with open(file_path, "r") as f:
config_data = yaml.safe_load(f)
return AppConfig(
database=DatabaseConfig(**config_data["database"]),
keydb=KeyDBConfig(**config_data["keydb"]),
jwt=JWTConfig(**config_data["jwt"]),
)
def configure() -> AppConfig:
title = "LitestarCats"
parser = argparse.ArgumentParser(title)
src_dir = Path(__file__).absolute().parent
config_file = src_dir / "settings-example.yaml"
parser.add_argument(
"-c", "--config", type=str, default=config_file, help="Config file"
)
args = parser.parse_known_args()
if args and args[0].config:
config_file = args[0].config
config = load_config(config_file)
return configmsgspec не поддерживает автоматическую десериализацию из словарей так, как это делает pydantic, поэтому мы вручную преобразуем данные из YAML в наши структуры. Это немного более явный подход, но он соответствует философии msgspec: быть лёгким и быстрым.
Создаём settings-example.yaml
Создаём файл settings-example.yaml с такой структурой, чтобы она соответствовала нашим классам:
database:
host: 127.0.0.1
port: 5432
user: postgres
password: postgres
database_name: ltcats_test_db
#url: "postgresql+asyncpg://postgres:postgres@localhost/ltcats_test_db" local
url: "postgresql+asyncpg://postgres:postgres@localhost/ltcats_test_db" # docker
echo: true
keydb:
host: 127.0.0.1
port: 6379
db: 0
url: "keydb://localhost:6379/0"
jwt:
secret: "your-secret-key"
token_secret: "super-secret"📌 Примечание: в продакшене лучше хранить секреты в централизованном secrets manager (Vault, AWS/GCP/Azure Secret Manager), но для простоты мы используем YAML. Если захотите, можно добавить поддержку .env через os.getenv или другую библиотеку.
Обновляем app.py
Теперь обновим src/app.py, чтобы использовать наши конфиги и добавим логирование, чтобы видеть что происходит во время работы приложения:
from litestar.contrib.sqlalchemy.plugins import (
AsyncSessionConfig,
SQLAlchemyAsyncConfig,
SQLAlchemyInitPlugin,
)
from litestar.logging import LoggingConfig
from src.configs.app_config import configure
# Загружаем конфиги
config = configure()
logging_config = LoggingConfig(
root={"level": "INFO", "handlers": ["queue_listener"]},
formatters={
"standard": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"}
},
log_exceptions="always", # Включить логирование исключений с трассировкой
)
...
session_config = AsyncSessionConfig(expire_on_commit=False)
db_config = SQLAlchemyAsyncConfig(
connection_string=config.database.get_connection_url(), # тут достаем переменную через config
before_send_handler="autocommit",
session_config=session_config,
)
app = Litestar(
route_handlers=[...],
logging_config=logging_config, # добавляем логирование
)Проверка
Запускаем приложение:
make run
Если всё работает, значит, конфиги подгружаются корректно. Теперь у нас есть единое место для всех настроек, и мы используем msgspec для консистентности с остальным проектом — котики довольны, порядок наведён! 🐾
Аутентификация с JWT — «Усы, лапы и хвост — вот мои документы!» 🔑
Зачем это нужно?
Безопасность — первое правило кошачьего клуба. Мы не хотим, чтобы Барсик редактировал резюме Мурзика без спроса! Для этого добавим аутентификацию через JWT, причём используем встроенный модуль из Litestar — быстро, надёжно и без лишних зависимостей. Только котики с пропуском смогут войти! 😺
Что будем делать?
Настроим JWT через JWTAuth из Litestar.
Добавим эндпоинт /login и защитим CRUD-операции.
Обновим модель User для хранения пароля.
Детали реализации
Добавляем пароль в модель
Сначала добавим поле для хранения хешированного пароля в нашу модель User. Открываем src/postgres/models/user.py и обновляем:
from sqlalchemy import String
from sqlalchemy import CheckConstraint, Index
from sqlalchemy.orm import Mapped, mapped_column, relationship
from litestar.plugins.sqlalchemy import base
from typing import Optional
class User(base.UUIDAuditBase):
tablename = "users"
first_name: Mapped[str] = mapped_column(String(50), nullable=False)
last_name: Mapped[str] = mapped_column(String(50), nullable=True)
email: Mapped[str] = mapped_column(
String(255), nullable=False, unique=True, index=True
)
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False) #добавлен пароль
profile_photo_url: Mapped[Optional[str]] = mapped_column(String(255))
user_roles: Mapped[list["UserRole"]] = relationship(back_populates="user")
cvs: Mapped[list["CV"]] = relationship(back_populates="user")
__table_args__ = (
CheckConstraint("length(first_name) > 0", name="check_first_name_not_empty"),
CheckConstraint("length(last_name) > 0", name="check_last_name_not_empty"),
CheckConstraint("email LIKE '%@%.%'", name="check_email_format"),
Index("idx_users_created_at", "created_at"),
)Теперь нужно сгенерировать миграцию, чтобы база данных узнала о новом поле:
make revision msg="add hashed_password to users"
make upgradeЭндпоинт логина и защита CRUD
Для хеширования паролей нам понадобится passlib, так что установим его:
uv add "passlib[bcrypt]"
Обновим src/controllers/user.py:
from litestar.di import Provide
from src.models.users import UserLogin
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["sha256_crypt"])
class UserController(Controller):
path = "/users"
dependencies = {
"users_repo": Provide(provide_users_repo),
}
@post("/login", signature_types=[User]) # Отключаем защиту для логина
async def login(
self,
data: UserLogin,
users_repo: UsersRepository,
) -> dict:
user = await users_repo.get_one_or_none(email=data.email)
if not user or not pwd_context.verify(data.password, user.hashed_password):
raise HTTPException(status_code=401, detail="Неверный email или пароль")
token = app.jwt_auth.create_token(identifier=str(user.id))
return {"access_token": token, "token_type": "bearer"}
Проверка
Давай проверим, как это работает. Сначала создаём пользователя (для теста можно временно убрать защиту с /users):
curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"first_name": "Мурзик", "last_name": "Котов", "email": "murzik@example.com"}'Теперь логинимся через /login:
curl -X POST http://localhost:8000/users/login -H "Content-Type: application/json" -d '{"first_name": "Мурзик", "last_name": "Котов", "email": "murzik@example.com"}'Получаем токен в ответе, например:
{
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
"token_type": "bearer"
}Используем токен для доступа к защищённым эндпоинтам:
curl -X GET http://localhost:8000/users/<user_id> -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."Если всё работает, то только котики с кошачьим пропуском могут войти! JWT от Litestar — это как миска с кормом: просто, вкусно и безопасно. 🐾
Кеширование с KeyDB — кошачья скорость ⚡
Зачем это нужно?
Если котики начнут массово запрашивать резюме, наша база скажет: «Мяу, я устала!» KeyDB с его многопоточными лапками ускорит всё, чтобы котики летали, а база отдыхала. Мы обещали использовать KeyDB ещё в первой части, так что пора выполнять обещания! 😺
Что будем делать?
Установим KeyDB через Docker.
Добавим кэширование для эндпоинта /users/{user_id}.
Детали реализации
Запуск KeyDB
Поднимаем KeyDB в Docker:
docker run -d -p 6379:6379 eqalpha/keydb
Добавляем зависимость
Устанавливаем библиотеку для работы с KeyDB:
uv add "redis[hiredis]"
Настройка клиента
Создаём файл src/clients/cache.py для работы с KeyDB, используя наш конфиг:
import redis.asyncio as redis
from src.configs.app_config import configure
config = configure()
keydb = redis.Redis(host=config.keydb.host, port=config.keydb.port, db=config.keydb.db)Кэширование в контроллере
Обновим метод get_user в src/controllers/user.py, чтобы он сначала проверял кеш, а только потом лез в базу. Добавим также эндпоинт для резюме (о нём позже):
from src.clients.cache import keydb
from litestar.response import Template
logger = logging.getLogger(__name__)
pwd_context = CryptContext(schemes=["sha256_crypt"])
class UserController(Controller):
path = "/users"
dependencies = {
"users_repo": Provide(provide_users_repo),
}
@get("/{user_id:uuid}", return_dto=MsgspecDTO[UserRead])
async def get_user(
self,
user_id: UUID,
users_repo: UsersRepository,
) -> UserRead:
"""Get an existing author."""
cache_key = f"user:{user_id}"
cached = await keydb.get(cache_key)
if cached:
return json.decode(cached, type=UserRead)
user = await users_repo.get(user_id)
await keydb.set(cache_key, json.encode(user.to_dict()), ex=3600) # Кэш на час
return user
# -------------------------------------
@get("/{user_id:uuid}/cv", media_type="text/html")
async def get_user_resume(self, user_id: UUID, users_repo: UsersRepository) -> Template:
user = await users_repo.get(user_id)
template = app.env.get_template("cv.html")
return Template(template=template, context={"user": user})Проверка
Запрашиваем пользователя через curl:
curl -X GET http://localhost:8000/users/<user_id> -H "Authorization: Bearer <your-token>"
Первый запрос пойдёт в базу, а второй — уже из кеша, и будет быстрее. KeyDB — это как кошачья мята для API: котики летают, а база отдыхает. Мяу-скорость включена! ⚡
Покрытие тестами с Coverage — проверяем кошачью надёжность 🧪
Зачем это нужно?
У нас есть тесты, но как понять, всё ли мы проверили? Coverage покажет, где котики ещё не прошлись лапками, и поможет убедиться, что наш код надёжен как кошачья интуиция. 😸
Что будем делать?
Установим coverage.
Настроим запуск тестов с покрытием и выведем отчёт.
Детали реализации
Установка
Устанавливаем coverage:
uv add coverage
Настройка
Обновим Makefile, чтобы запускать тесты с покрытием:
test-coverage:
uv run coverage run -m pytest
test-coverage-report:
uv run coverage report --show-missingЗапуск
Запускаем:
make test-coverage-report
Пример вывода:
Name Stmts Miss Cover Missing
-----------------------------------------------------
src/app.py 20 2 90% 15-16
src/controllers/users.py 50 10 80% 25-30, 45-50
src/models/users.py 10 0 100%
-----------------------------------------------------
TOTAL 80 12 85%Проверка
Отчёт показывает, что у нас 85% покрытия — неплохо, но есть куда расти! Например, строки 25-30 в users.py — это обработка ошибок в /login, которую мы не протестировали. Давай добавим тест в src/tests/test_users.py:
@pytest.mark.asyncio
async def test_login_invalid_credentials(db_session: AsyncSession, test_client):
# 1. Создаем пользователя с некорректным паролем
user_data = UserCreate(
first_name="Васька",
last_name="Мурзиков",
email="vasyamurzikov@whiskers.com",
password="12345",
)
user_dict = structs.asdict(user_data)
password = user_dict.pop("password")
user = User(**user_dict, hashed_password=pwd_context.hash("wrong-pass"))
db_session.add(user)
await db_session.commit()
# Данные для логина
login_data = UserLogin(email=user.email, password=password)
# Отправка POST-запроса через тестовый клиент
response = await test_client.post("/users/login", json=structs.asdict(login_data))
# Проверка результата
assert response.status_code == 401
assert response.json().get("detail") == "Неверный email или пароль"И фикстуру тестового клиента в conftest.py:
from src.app import app
from litestar.testing import AsyncTestClient
@pytest_asyncio.fixture
async def test_client(db_session: AsyncSession) -> AsyncTestClient:
async with AsyncTestClient(app) as client:
yield clientТеперь повторяем make test-coverage — покрытие должно подрасти! Coverage — это как ветеринар для кода: сразу видно, где котик прихрамывает. Пора подтянуть хвосты! 🐾
Шаблон резюме с Jinja — кошачьи карточки
Зачем это нужно?
API — это круто, но хочется увидеть резюме котиков вживую! Сделаем шаблон в стиле баскетбольных карточек 90-х: звёзды сверху, имя-фамилия, описание, данные и табличка с опытом работы. Минимум JS и CSS, только чистый кошачий шик! 😺
Что будем делать?
Установим jinja2.
Добавим модель и эндпоинт для отображения резюме (уже добавили в users.py).
Создадим HTML-шаблон в стиле баскетбольных карточек.
Детали реализации
Установка Jinja
Устанавливаем jinja2:
uv add jinja2
Настройка Jinja
Добавим в файл src/app.py код для работы с шаблонами:
from jinja2 import Environment, PackageLoader, select_autoescape
env = Environment(
loader=PackageLoader("src"),
autoescape=select_autoescape()
)Шаблон cv.html
Шрифт на карточке похож на Impact для заголовков и Arial для текста. Мы убрали фото, добавили звёзды, описание и табличку с опытом работы. Создаём templates/cv.html:
<!DOCTYPE html>
<html>
<head>
<title>{{ user.first_name }} {{ user.last_name }} - Resume</title>
<style>
.card {
border: 2px solid #000;
width: 400px;
margin: 20px auto;
background: #f0f0f0;
font-family: Arial, sans-serif;
box-shadow: 5px 5px 10px rgba(0, 0, 0, 0.3);
}
.stars {
text-align: center;
font-size: 24px;
color: #ff4500;
padding: 5px;
}
.header {
background: #ff4500;
color: white;
text-align: center;
padding: 10px;
font-family: Impact, sans-serif;
font-size: 24px;
text-transform: uppercase;
}
.subheader {
font-family: Impact, sans-serif;
font-size: 14px;
color: #ff4500;
text-align: center;
margin: 5px 0;
}
.info {
padding: 10px;
font-size: 14px;
}
.description {
font-style: italic;
margin: 10px 0;
}
.stats {
margin: 10px 0;
}
.stats div {
margin: 5px 0;
}
.experience {
border-top: 2px solid #ff4500;
padding: 10px;
}
.experience h3 {
font-family: Impact, sans-serif;
font-size: 16px;
text-align: center;
margin-bottom: 10px;
color: #000;
}
.experience table {
width: 100%;
border-collapse: collapse;
font-size: 12px;
}
.experience th, .experience td {
border: 1px solid #000;
padding: 5px;
text-align: left;
}
.experience th {
background: #ff4500;
color: white;
font-family: Impact, sans-serif;
}
</style>
</head>
<body>
<div class="card">
<div class="stars">⭐⭐⭐⭐⭐</div>
<div class="header">{{ user.first_name }} {{ user.last_name }}</div>
<div class="subheader">PROFESSIONAL CAT CODER</div>
<div class="info">
<div class="stats">
<div><b>Email:</b> {{ user.email }}</div>
<div><b>Joined:</b> {{ user.created_at.strftime('%Y-%m-%d') }}</div>
</div>
<div class="description">
{{ user.first_name }} is a purr-fect coder with a knack for catching bugs faster than a laser pointer. Known for napping on keyboards and delivering meow-nificent code, this cat is a true asset to any team!
</div>
</div>
<div class="experience">
<h3>WORK EXPERIENCE</h3>
<table>
<tr>
<th>Компания</th>
<th>Должность</th>
<th>Тип работы</th>
<th>Начало работы</th>
<th>Конец работы</th>
<th>Описание</th>
</tr>
{% for cv in user.cvs %}
{%- for exp in cv.work_experiences %}
<tr>
<td>{{ exp.company }}</td>
<td>{{ exp.job_title }}</td>
<td>{{ exp.employment_type }}</td>
<td>{{ exp.start_date }}</td>
<td>{{ exp.end_date }}</td>
<td>{{ exp.description }}</td>
</tr>
{% endfor %}
{% endfor %}
</table>
</div>
</div>
</body>
</html>
Проверка
Добавим тестовые данные через SQL или API, чтобы у пользователя был опыт работы. Например:
INSERT INTO work_experience (user_id, company, position, description, created_at, updated_at)
VALUES ('<user_id>', 'CatTech', 'Senior Whisker Engineer', 'Python, MeowSQL', NOW(), NOW());Теперь открываем в браузере http://localhost:8000/users/<user_id>/cv. Вы увидите стильную карточку: звёзды сверху, имя-фамилия, описание и табличка с опытом работы. Без лишнего JS — только чистый кошачий шик!
Упаковка в Docker — котики в коробке 📦
Зачем это нужно?
В продакшене код без Docker — как котик без коробки. Упакуем всё аккуратно, чтобы наш API был готов к деплою! 📦
Что будем делать?
Создадим Dockerfile.
Обновим docker-compose.yaml.
Детали реализации
Dockerfile
Создаём Dockerfile:
FROM python:3.13.3-slim
WORKDIR /app
# Копируем файлы с зависимостями и конфиг
COPY pyproject.toml uv.lock ./
COPY src/configs/settings-example.yaml ./configs/
# Устанавливаем uv (если он не установлен в базовом образе)
RUN pip install --no-cache-dir uv
# Устанавливаем зависимости через uv
RUN uv pip install --system -r pyproject.toml
# Копируем исходный код и остальные файлы
COPY src/ ./src/
COPY src/templates/ ./templates/
COPY src/configs/alembic.ini ./configs/
# Запускаем приложение через granian
CMD ["granian", "--interface", "asgi", "src.app:app"]Обновлённый docker-compose.yaml
Обновляем docker-compose.yaml, чтобы включить KeyDB и приложение:
services:
app:
build: .
ports:
- "8000:8000"
depends_on:
postgres:
condition: service_healthy
keydb:
condition: service_healthy
networks:
- litestarcats
postgres:
image: postgres:latest
container_name: postgres
hostname: postgres
ports:
- 5432:5432
volumes:
- "postgres-data:/var/lib/postgresql/data"
networks:
- litestarcats
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 10s
timeout: 5s
retries: 3
keydb:
image: eqalpha/keydb:latest
container_name: keydb
ports:
- "6379:6379"
volumes:
- keydb_data:/data
restart: unless-stopped
networks:
- litestarcats
healthcheck:
test: ["CMD-SHELL", "redis-cli ping"]
interval: 10s
timeout: 5s
retries: 3
volumes:
postgres-data:
keydb_data:
driver: local
networks:
litestarcats:
driver: bridgeПроверка
Запускаем:
docker compose up --build
Проверяем, что API доступно на http://localhost:8000 и резюме отображается. Котики в коробке, а проект в продакшене! Docker — это как переноска для нашего API. 🐾
Котики готовы к продакшену! 🎉
Итоги
Мы вынесли конфиги в отдельный модуль с помощью msgspec, добавили JWT для безопасности, KeyDB для скорости, coverage для надёжности, Jinja для стиля и Docker для деплоя. Наш кошачий API теперь готов к бою! 😺
Впечатления
Litestar — это находка, KeyDB летает, а msgspec сделал код ещё быстрее и консистентнее. Карточки получились на ура, Мурзик теперь выглядит как звезда! ⭐
Что дальше?
В следующей части можно Litestar с FastAPI. А может, кошачьи аватарки через API? Пишите в комментариях, что хотите в четвёртой части! 🐾
Ссылка
Код на GitHub. Лапки вверх, если понравилось! 😻
P.S.: Если заметили ошибку, напишите мне пожалуйста, я исправлю. Код на гитхабе обновлён, там всегда актуальная, рабочая версия, если тут вдруг стало не понятно или нужно посмотреть код целиком, можете посмотреть там или опять же, написать мне, я постараюсь поправить в статье, так чтобы стало лучше. Спасибо.