Котики выходят на новый уровень! 🐾
Привет, котики и котолюбы! В первой части нашего кошачьего приключения мы выбрали инструменты (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 import Litestar from litestar.di import Provide from src.controllers.user import UserController, provide_users_repo from src.controllers.role import RoleController from src.controllers.user_role import UserRoleController from src.controllers.cv import CVController from src.controllers.work_experience import WorkExperienceController from src.controllers.company import CompanyController from src.controllers.educational_institution import EducationalInstitutionController from src.controllers.education import EducationController from litestar.contrib.sqlalchemy.plugins import ( AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin, ) from litestar.connection import ASGIConnection from litestar.openapi import OpenAPIConfig from litestar.params import Parameter from litestar.plugins.sqlalchemy import filters, base from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin from litestar.security.jwt import JWTAuth, Token from litestar.logging import LoggingConfig from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from uuid import UUID from src.configs.app_config import configure from src.postgres.models.user import User from jinja2 import Environment, PackageLoader, select_autoescape # Загружаем конфиги 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", # Включить логирование исключений с трассировкой ) env = Environment( loader=PackageLoader("src"), autoescape=select_autoescape() ) async def provide_limit_offset_pagination( current_page: int = Parameter(ge=1, query="currentPage", default=1, required=False), page_size: int = Parameter( query="pageSize", ge=1, default=10, required=False, ), ) -> filters.LimitOffset: """Add offset/limit pagination. Return type consumed by Repository.apply_limit_offset_pagination(). Parameters ---------- current_page : int LIMIT to apply to select. page_size : int OFFSET to apply to select. """ return filters.LimitOffset(page_size, page_size * (current_page - 1)) sessionmaker = async_sessionmaker(expire_on_commit=False) async def retrieve_user_handler( token: Token, connection: ASGIConnection, ) -> User | None: user_id = UUID(token.sub) users_repo = connection.scope.get("users_repo") if not users_repo: async with sessionmaker(bind=db_config.get_engine()) as session: try: async with session.begin(): users_repo = await provide_users_repo(db_session=session) except IntegrityError as exc: raise ClientException( status_code=HTTP_409_CONFLICT, detail=str(exc), ) from exc user = await users_repo.get(user_id) return user jwt_auth = JWTAuth[User]( retrieve_user_handler=retrieve_user_handler, token_secret=config.jwt.token_secret, algorithm="HS256", exclude=["/users", "/schema"] ) session_config = AsyncSessionConfig(expire_on_commit=False) db_config = SQLAlchemyAsyncConfig( connection_string=config.database.get_connection_url(), before_send_handler="autocommit", session_config=session_config, ) async def on_startup() -> None: """Initializes the database.""" async with db_config.get_engine().begin() as conn: await conn.run_sync(base.UUIDBase.metadata.create_all) sqlalchemy_plugin = SQLAlchemyInitPlugin(config=db_config) app = Litestar( route_handlers=[ UserController, RoleController, UserRoleController, CVController, WorkExperienceController, CompanyController, EducationalInstitutionController, EducationController, ], on_startup=[on_startup], on_app_init=[jwt_auth.on_app_init], openapi_config=OpenAPIConfig(title="My API", version="1.0.0"), dependencies={ "limit_offset": Provide(provide_limit_offset_pagination), }, plugins=[sqlalchemy_plugin, SQLAlchemySerializationPlugin()], 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 UsersRepository(repository.SQLAlchemyAsyncRepository[User]): """Author repository.""" model_type = User async def provide_users_repo(db_session: AsyncSession) -> UsersRepository: """This provides the default Authors repository.""" return UsersRepository(session=db_session) 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): user_data = UserCreate(first_name="Васька", last_name="Мурзиков", email="vasya@whiskers.com") user = User(**structs.asdict(user_data), hashed_password=pwd_context.hash("wrong-pass")) db_session.add(user) await db_session.commit() with pytest.raises(HTTPException) as exc: await UserController().login(user_data, UsersRepository(session=db_session)) assert exc.value.status_code == 401Теперь повторяем 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>Company</th> <th>Position</th> <th>Tech Stack</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. Лапки вверх, если понравилось! 😻
ссылка на оригинал статьи https://habr.com/ru/articles/901852/
Добавить комментарий