Как стать автором
Поиск
Написать публикацию
Обновить
57.25
Рейтинг
NtechLab
Мировой лидер в разработке решений на основе ИИ.

LitestarCatsCV. Тренируемся на кошках. Расширяем возможности и готовимся к продакшену. Часть 3

Уровень сложностиПростой
Время на прочтение17 мин
Количество просмотров631
Блог компании NtechLabPython*API*Программирование*Веб-разработка*
Туториал

Котики выходят на новый уровень! 🐾

Привет, котики и котолюбы! В первой части нашего кошачьего приключения мы выбрали инструменты (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 config

msgspec не поддерживает автоматическую десериализацию из словарей так, как это делает 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.: Если заметили ошибку, напишите мне пожалуйста, я исправлю. Код на гитхабе обновлён, там всегда актуальная, рабочая версия, если тут вдруг стало не понятно или нужно посмотреть код целиком, можете посмотреть там или опять же, написать мне, я постараюсь поправить в статье, так чтобы стало лучше. Спасибо.

Теги:
Хабы:
Если эта публикация вас вдохновила и вы хотите поддержать автора — не стесняйтесь нажать на кнопку
+4
Комментарии0
NtechLab
Мировой лидер в разработке решений на основе ИИ.
Сайт
4
Карма
6
Рейтинг
Александр @pulichkin

Бэкенд разработчик python

Отправить сообщение

Публикации

Информация

Сайт
ntechlab.ru
Дата регистрации
Дата основания
Численность
201–500 человек
Местоположение
Россия
Представитель
Eli_bas

Ваш аккаунт

Разделы

Информация

Услуги

Техническая поддержка
© 2006–2025, Habr