LitestarCatsCV. Тренируемся на кошках. Реализация API и работа с данными. Часть 2

В этой статье вы узнаете, как:

• Поднять PostgreSQL и подготовить базу данных.

• Настроить тесты с Pytest и внедрить подход TDD.

• Создать модели данных и миграции с SQLAlchemy и Alembic.

• Реализовать CRUD-операции с помощью Litestar.

• Проверить работу API с использованием curl.

---

Что вас ждёт:
Если в первой части мы заложили фундамент проекта (выбор инструментов, настройка окружения и структура), то здесь мы превратим этот каркас в полноценное API для управления резюме кошек (или людей — как вам ближе). Мы подключим базу данных, добавим тесты, настроим миграции и даже проверим всё в действии. К концу статьи у вас будет рабочее API, которое можно потрогать руками (или лапками 🐾). Полный код доступен на GitHub — ссылка в конце!

Вступление: с чего начнём?

В первой части мы остановились на базовой структуре проекта и настройке зависимостей.
А именно:

• Выбрали инструменты — Litestar вместо Fastapi, Granian вместо Uvicorn, KeyDB вместо Redis.

• Настроили окружение с помощью uv.

• Создали структуру проекта.

• Добавили Makefile для удобства работы.

Теперь мы готовы двигаться дальше и строить “стены” нашего приложения. Сегодня мы превратим заготовку в рабочее API с CRUD-операциями, подключим базу через SQLAlchemy, настроим миграции с Alembic и напишем первые тесты с Pytest. Поехали! 🚀

Подготовка PostgreSQL

Если у вас уже есть PostgreSQL, просто создайте новую базу данных. Если нет — давайте поднимем его в Docker. Это быстро и удобно!

🚀 Шаг 1: Создание docker-compose файла

Создаём файл:

nvim docker-compose.yaml

И вставляем туда этот код:

services:
  postgres:
    image: postgres
    container_name: postgres
    hostname: postgres
    ports:
      - 5432:5432
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: ltcats_test_db
    volumes:
      - "postgres-data:/var/lib/postgesql"
volumes:
  postgres-data:

Запускаем контейнер:

docker compose up -d

Проверяем, что всё работает:

docker ps --format 'table {{.ID}}\t{{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}'

Ожидаемый вывод:

CONTAINER ID   NAMES            IMAGE                                                                   
STATUS          PORTS
2099ccd4d2dd   postgres         postgres                                                                
Up 22 seconds   0.0.0.0:5432->5432/tcp, [::]:5432->5432/tcp

🧪 Шаг 2: Проверка подключения

Убедимся, что база доступна. Используйте psql:

psql -h localhost -p 5432 -U postgres -d ltcats_test_db

Или, если у вас есть pgcli:

pgcli -h localhost -p 5432 -U postgres -d ltcats_test_db

📦 Шаг 3: Добавляем зависимости

База готова, пора подключить её к проекту. Добавляем библиотеки через uv:

uv add asyncpg litestar-asyncpg

✅ PostgreSQL готов к работе!

Настройка тестов с Pytest

Начинаем с тестов, это наш первый шаг к надёжному коду. Мы будем использовать подход TDD (Test-Driven Development), чтобы сразу проверять, что всё работает как надо.

🧪 Шаг 1: Установка зависимостей

Добавляем нужные пакеты:

uv add pytest pytest-asyncio sqlalchemy advanced-alchemy

📂 Шаг 2: Создание структуры для тестов

Создаём файлы в папке src/tests:

mkdir -p src/tests && touch src/tests/__init__.py src/tests/conftest.py src/tests/test_users.py

Теперь структура проекта выглядит так:

.
├── Makefile
├── README.md
├── docker-compose.yaml
├── pyproject.toml
├── src
│   ├── app.py
│   └── tests
│       ├── __init__.py,
│       ├── conftest.py,
│       └── test_users.py
└── uv.lock

⚙️ Шаг 3: Настройка Pytest

В pyproject.toml добавляем конфигурацию:

[tool.pytest.ini_options]
asyncio_default_fixture_loop_scope = "function"

🛠️ Шаг 4: Фикстуры в conftest.py

Файл conftest.py нужен для:

• Хранения общих фикстур, которые используются в тестах. Фикстуры — это функции, которые выполняют настройку и очистку данных для тестов.

• Избегания повторяющегося кода: Помогает избегать дублирования кода в тестах. Если есть функции или данные, которые используются в разных тестах, их можно вынести в conftest.py.

• Упрощения тестов: Сделать тесты более читаемыми и простыми, вынося общую логику в отдельный файл.

Вот его содержимое:

import asyncio
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from postgres.models.base import Base

# Настройка тестовой БД
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/ltcats_test_db"

# Создаем новый событийный цикл, который будет использоваться для запуска асинхронных тестов.
@pytest_asyncio.fixture(scope="session") # Параметр scope="session" означает, что событийный цикл будет создан один раз на всю сессию тестирования, а не для каждого теста отдельно
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop

# Фикстура для движка базы данных
@pytest_asyncio.fixture(scope="session")
async def db_engine():
    # Создает движок с подключением к тестовой базе данных
    engine = create_async_engine(TEST_DATABASE_URL, echo=True) # Параметр echo=True включает вывод SQL-запросов в консоль для отладки.
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all) #Создает все таблицы в базе данных с помощью Base.metadata.create_all.
    yield engine # Выдает движок для использования в тестах.
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all) # После завершения тестов удаляет все таблицы (Base.metadata.drop_all)
    await engine.dispose() # и освобождает ресурсы движка (engine.dispose()).

# Фикстура для сессии базы данных
@pytest_asyncio.fixture
async def db_session(db_engine) -> AsyncSession:
    async with AsyncSession(db_engine) as session: # Создает новую сессию базы данных.
        yield session # Выдает сессию для использования в тестах.
        await session.rollback() # После завершения теста откатывает все изменения (session.rollback()), чтобы база данных оставалась в исходном состоянии.

✍️ Шаг 5: Пишем первый тест

В test_users.py добавляем тест для создания пользователя
Далее перейдём в test_users.py и так как в нашем приложении будут пользователи, то первым делом мы займёмся ими.
Определим:

• Как будет выглядеть структура создания пользователя.

• Как будет инициализироваться модель.

• Добавим пользователя в базу.

• Достанем его из базы и проверим, сохранились ли данные.

Вперёд, напишем первый тест:

import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models.users import UserCreate
from postgres.models.users import User
from msgspec import structs


# Маркируем тест как асинхронный
@pytest.mark.asyncio
async def test_create_user(db_session: AsyncSession):
    # Создаем данные для нового пользователя (котика)
    user_data = UserCreate(
        first_name="Васька",  # Имя котика
        last_name="Мурзиков",  # Фамилия котика
        email="[e-mail]",  # Почта котика
    )
    # Создаем объект пользователя из данных
    user = User(**structs.asdict(user_data))
    # Добавляем пользователя в сессию базы данных
    db_session.add(user)
    # Сохраняем изменения в базе данных
    await db_session.commit()
    # Выполняем запрос для получения пользователя по email
    result = await db_session.execute(
        select(User).where(User.email == "[e-mail]")
    )
    # Получаем пользователя из результата запроса
    fetched_user = result.scalar_one()
    # Проверяем, что имя и email пользователя соответствуют ожидаемым значениям
    assert fetched_user.first_name == "Васька"
    assert fetched_user.email == "[e-mail]"

🚀 Шаг 6: Запускаем тесты

Запускаем командой:

uv run pytest

Или добавляем в Makefile:

test:
    uv run pytest

И запускаем через make:

make test

После чего, закономерно узнаем, что наш тест упал:

ERROR src/tests - ModuleNotFoundError: No module named 'postgres'

Тест падает с ошибкой ModuleNotFoundError, значит, мы забыли создать модели. Переходим к следующему шагу!

Модели данных и миграции

Теперь определим модели и настроим миграции, чтобы база данных соответствовала нашему коду.

📂 Шаг 1: Создание файлов моделей

Создаём структуру:

mkdir -p src/models src/postgres/models && \
touch src/models/__init__.py src/postgres/__init__.py src/postgres/models/__init__.py \
src/models/users.py src/postgres/models/users.py

Проверяем, структуру и файлы:

src
 ├── models
 │   ├── __init__.py
 │   └── users.py
 ├── postgres
 │   ├── __init__.py
 │   └── models
 │       ├── __init__.py
 │       └── users.py

Переходим к следующему шагу.

Определяем модели с использованием SQLAlchemy:
📌 Примечание: Тут надо сделать небольшую сноску, я планирую использовать в проекте контроллер и репозиторий, поэтому классы буду наследовать от base.UUIDAuditBase и возможно от base.UUIDBase, для этого сценария подходят только они. Я потратил некоторое количество времени, чтобы убедиться в этом, а если хотите пойти другим путём, то в целом можете пользоваться любыми другими.
И ещё при наследовании от base.UUIDBase в таблице автоматически добавится поле id с uuid’ом, а если наследоваться от base.UUIDAuditBase, то помимо id добавятся такие поля, как created_at и updated_at, надо иметь это ввиду.

📋 Шаг 2: Модель User

В src/postgres/models/users.py добавляем:

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from litestar.plugins.sqlalchemy import base


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)

📦 Шаг 3: Добавляем msgspec

Здесь пора добавить в проект msgspec:

uv add msgspec

✍️ Шаг 4: Структура UserCreate

И обновим файл src/models/users.py:

from msgspec import Struct


class UserCreate(Struct, kw_only=True, omit_defaults=True):
    first_name: str
    last_name: str | None = None
    email: str

Эта структура, в последствии, понадобиться для удобства и валидации входных данных.
Параметр kw_only=True в определении структуры (msgspec.Struct), означает, что все поля структуры должны передаваться, как ключевые аргументы, при создании экземпляра класса.

✅ Шаг 5: Проверяем тест

Запускаем make test. Теперь тест должен пройти успешно!
Ожидаемый вывод:

❯ uv run pytest
======================================================================= test session starts =======================================================================
platform linux -- Python 3.13.1, pytest-8.3.4, pluggy-1.5.0
rootdir: /home/user/uprojectfolder/litestarcatscv
configfile: pyproject.toml
plugins: Faker-35.2.0, anyio-4.8.0, asyncio-0.25.3
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=function
collected 1 item

src/tests/test_users.py .                                                                                                                                   [100%]

======================================================================== 1 passed in 0.24s ========================================================================

Настройка миграций с Alembic

Пора синхронизировать наши модели с базой данных через миграции.

🔧 Шаг 1: Установка Alembic

Добавляем в проект alembic:

uv add alembic

⚙️ Шаг 2: Инициализация Alembic

📌 Примечание: В версии SQLAlchemy 1.4 появилась экспериментальная поддержка asyncio, позволяющая использовать большую часть интерфейса в асинхронных приложениях. Alembic в настоящее время не предоставляет асинхронный API напрямую, но может использовать движок SQLAlchemy Async для запуска миграций и автоматической генерации.

Новые конфигурации могут использовать шаблон -t «async» для запуска среды, которую можно использовать с асинхронным DBAPI, например asyncpg, выполнив команду:

Создаём структуру для миграций:

alembic init -t async src/postgres/alembic/

📂 Шаг 3: Переносим конфигурацию

Переносим alembic.ini в папку src/configs:

mkdir -p src/configs && touch src/configs/__init__.py
mv alembic.ini src/configs/

⚙️ Шаг 4: Настраиваем alembic.ini

Обновляем файл:

script_location = src/postgres/alembic
version_locations = src/postgres/alembic/versions
sqlalchemy.url = postgresql+asyncpg://postgres:postgres@localhost/ltcats_test_db

🛠️ Шаг 5: Настраиваем env.py

В src/postgres/alembic/env.py добавляем:

import importlib
import os
from src.postgres.models.base import Base


# Настройка конфигурации
config = context.config
config_file = os.path.abspath(os.path.join(os.path.dirname(__file__), '../..', 'configs', 'alembic.ini'))
fileConfig(config_file)

# Динамически импортируем все модели из src/models/
models_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'models'))
for filename in os.listdir(models_dir):
    if filename.endswith('.py') and filename not in ['__init__.py', 'base.py']:
        module_name = f"src.postgres.models.{filename[:-3]}"
        importlib.import_module(module_name)

#Добавляем Метаданные модели
target_metadata = Base.metadata

🚀 Шаг 6: Создаём первую миграцию

Генерируем миграцию:

alembic -c src/configs/alembic.ini revision -m "create user table" --autogenerate

Ожидаемый вывод:

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'users'
INFO  [alembic.autogenerate.compare] Detected added index ''ix_users_email'' on '('email',)'
  Generating /home/user/uprojects/litestarcatscv/src/postgres/alembic/versions/9047a811291c_create_user_table.py ...  done

А в папке src/postgres/alembic/versions, появиться новый файл миграции со следующим содержимым:

"""create user table

Revision ID: 2d7f1d2c604d
Revises:
Create Date: YYYY-MM-DD

"""
import advanced_alchemy
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '2d7f1d2c604d'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('first_name', sa.String(length=50), nullable=False),
    sa.Column('last_name', sa.String(length=50), nullable=True),
    sa.Column('email', sa.String(length=255), nullable=False),
    sa.Column('id', advanced_alchemy.types.guid.GUID(length=16), nullable=False),
    sa.Column('sa_orm_sentinel', sa.Integer(), nullable=True),
    sa.Column('created_at', advanced_alchemy.types.datetime.DateTimeUTC(timezone=True), nullable=False),
    sa.Column('updated_at', advanced_alchemy.types.datetime.DateTimeUTC(timezone=True), nullable=False),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_users'))
    )
    op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_users_email'), table_name='users')
    op.drop_table('users')
    # ### end Alembic commands ###

✅ Шаг 7: Применяем миграцию

Применяем изменения:

alembic -c src/configs/alembic.ini upgrade head

Вывод:

Module --  src.postgres.models.users
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 9047a811291c, create user table

📋 Шаг 8: Обновляем Makefile

Добавляем удобные команды, чтобы было проще работать с alembic:

ALEMBIC_CONFIG = src/configs/alembic.ini

check-alembic:
	@command -v alembic >/dev/null 2>&1 || { echo "Alembic is not installed. Run 'make install'."; exit 1; }

revision: check-alembic
	alembic -c $(ALEMBIC_CONFIG) revision -m '$(msg)' --autogenerate

upgrade: check-alembic
	alembic -c $(ALEMBIC_CONFIG) upgrade head

downgrade: check-alembic
	alembic -c $(ALEMBIC_CONFIG) downgrade -1

Теперь миграции запускаются просто:

make revision msg="create tables"
make upgrade
make downgrade

Реализация API с Litestar

Пришло время создать CRUD-эндпоинты для работы с пользователями.

🌐 Шаг 1: Создаём контроллер

Для эндпоинтов будем использовать контроллеры.
Для этого создадим директорию controllers и добавим туда файлы:

mkdir -p src/controllers && touch src/controllers/__init__.py src/controllers/users.py

✍️ Шаг 2: Код контроллера

В src/controllers/users.py:

# Импортируем необходимые модули из Litestar для создания контроллера и обработки HTTP-запросов
from litestar import Controller, get, post, patch, delete
# Импортируем Provide для работы с зависимостями
from litestar.di import Provide
# Импортируем асинхронную сессию SQLAlchemy для работы с базой данных
from sqlalchemy.ext.asyncio import AsyncSession
# Импортируем схемы данных для создания и обновления пользователей
from src.models.users import UserCreate, UserPatch
# Импортируем модель пользователя для работы с базой данных
from src.postgres.models.users import User
# Импортируем утилиты msgspec для преобразования данных
from msgspec import structs, to_builtins
# Импортируем пагинацию из Litestar для вывода списков с ограничением и смещением
from litestar.pagination import OffsetPagination
# Импортируем фильтры и репозиторий из плагина SQLAlchemy для Litestar
from litestar.plugins.sqlalchemy import (
    filters,
    repository,
)
# Импортируем тип UUID для работы с уникальными идентификаторами
from uuid import UUID


# Определяем класс репозитория для работы с пользователями в базе данных
class UsersRepository(repository.SQLAlchemyAsyncRepository[User]):
    """Репозиторий для работы с моделью пользователей в базе данных."""
    model_type = User  # Указываем, что репозиторий работает с моделью User


# Функция-зависимость для предоставления экземпляра репозитория
async def provide_users_repo(db_session: AsyncSession) -> UsersRepository:
    """Создает и возвращает экземпляр UsersRepository с переданной сессией базы данных."""
    return UsersRepository(session=db_session)


# Определяем контроллер для управления пользователями
class UserController(Controller):
    path = "/users"  # Базовый путь для всех маршрутов контроллера (например, /users/)
    dependencies = {"users_repo": Provide(provide_users_repo)}  # Зависимость репозитория пользователей

    # Метод для получения списка пользователей с пагинацией
    @get(path="/")
    async def list_users(
        self,
        users_repo: UsersRepository,  # Репозиторий для доступа к данным
        limit_offset: filters.LimitOffset,  # Параметры пагинации (ограничение и смещение)
    ) -> OffsetPagination[User]:
        """Возвращает список пользователей с пагинацией."""
        # Получаем список пользователей и их общее количество из репозитория
        results, total = await users_repo.list_and_count(limit_offset)
        # Возвращаем объект пагинации с результатами
        return OffsetPagination[User](
            items=results,  # Список пользователей
            total=total,  # Общее количество пользователей
            limit=limit_offset.limit,  # Лимит записей на странице
            offset=limit_offset.offset,  # Смещение (сколько записей пропущено)
        )

    # Метод для создания нового пользователя
    @post("/")
    async def create_user(
        self,
        data: UserCreate,  # Данные для создания пользователя (схема UserCreate)
        users_repo: UsersRepository  # Репозиторий для работы с базой
    ) -> User:
        """Создает нового пользователя в базе данных."""
        # Преобразуем данные из схемы в словарь и создаем объект User
        user = await users_repo.add(User(**structs.asdict(data)))
        # Фиксируем изменения в базе данных
        await users_repo.session.commit()
        return user  # Возвращаем созданного пользователя

    # Метод для получения пользователя по его UUID
    @get("/{user_id:uuid}")
    async def get_user(
        self,
        user_id: UUID,  # Уникальный идентификатор пользователя
        users_repo: UsersRepository  # Репозиторий для доступа к данным
    ) -> User:
        """Возвращает данные пользователя по его UUID."""
        # Получаем пользователя из репозитория по идентификатору
        user = await users_repo.get(user_id)
        return user  # Возвращаем найденного пользователя

    # Метод для частичного обновления пользователя
    @patch("/{user_id:uuid}")
    async def update_user(
        self,
        user_id: UUID,  # Уникальный идентификатор пользователя
        data: UserPatch,  # Данные для обновления (схема UserPatch)
        users_repo: UsersRepository  # Репозиторий для работы с базой
    ) -> User:
        """Обновляет данные существующего пользователя."""
        # Преобразуем данные из схемы в словарь
        raw_obj = to_builtins(data)
        # Добавляем идентификатор пользователя в данные
        raw_obj["id"] = user_id
        # Создаем объект User с обновленными данными
        user = User(**raw_obj)
        # Обновляем пользователя в репозитории
        updated_user = await users_repo.update(user)
        # Фиксируем изменения в базе данных
        await users_repo.session.commit()
        return updated_user  # Возвращаем обновленного пользователя

    # Метод для удаления пользователя
    @delete("/{user_id:uuid}")
    async def delete_user(
        self,
        user_id: UUID,  # Уникальный идентификатор пользователя
        users_repo: UsersRepository  # Репозиторий для работы с базой
    ) -> None:
        """Удаляет пользователя из базы данных по его UUID."""
        # Удаляем пользователя из репозитория по идентификатору
        await users_repo.delete(user_id)
        # Фиксируем изменения в базе данных
        await users_repo.session.commit()

⚙️ Шаг 3: Обновляем app.py

В src/app.py:

# Импортируем необходимые модули из Litestar для создания приложения и работы с зависимостями
from litestar import Litestar
from litestar.di import Provide

# Импортируем контроллер пользователей из нашего проекта
from src.controllers.users import UserController

# Импортируем модули для асинхронной работы с базой данных через SQLAlchemy
from litestar.contrib.sqlalchemy.plugins import (
    AsyncSessionConfig,           # Конфигурация асинхронной сессии
    SQLAlchemyAsyncConfig,        # Основная конфигурация SQLAlchemy
    SQLAlchemyInitPlugin,         # Плагин для инициализации SQLAlchemy
)

# Импортируем Parameter для задания параметров запросов
from litestar.params import Parameter

# Импортируем фильтры и базовый класс моделей из плагина SQLAlchemy
from litestar.plugins.sqlalchemy import filters, base

# Импортируем плагин для сериализации моделей SQLAlchemy в JSON
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin

# Настройка базы данных
# Определяем строку подключения к базе данных PostgreSQL с использованием драйвера asyncpg
DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/ltcats_test_db"

# Функция для предоставления пагинации в запросах
async def provide_limit_offset_pagination(
    current_page: int = Parameter(
        ge=1,              # Значение должно быть больше или равно 1
        query="currentPage",  # Имя параметра в запросе
        default=1,         # Значение по умолчанию — первая страница
        required=False,    # Параметр необязательный
    ),
    page_size: int = Parameter(
        query="pageSize",  # Имя параметра в запросе
        ge=1,              # Значение должно быть больше или равно 1
        default=10,        # По умолчанию 10 элементов на странице
        required=False,    # Параметр необязательный
    ),
) -> filters.LimitOffset:
    """
    Предоставляет параметры пагинации для запросов к базе данных.
    Возвращает объект LimitOffset, который задает лимит и смещение для выборки данных.
    - current_page: номер текущей страницы.
    - page_size: количество элементов на странице.
    """
    return filters.LimitOffset(page_size, page_size * (current_page - 1))

# Настройка асинхронной сессии SQLAlchemy
# expire_on_commit=False предотвращает истечение объектов после коммита транзакции
session_config = AsyncSessionConfig(expire_on_commit=False)

# Конфигурация подключения SQLAlchemy к базе данных
db_config = SQLAlchemyAsyncConfig(
    connection_string=DATABASE_URL,    # Строка подключения к базе данных
    before_send_handler="autocommit",  # Автоматический коммит после выполнения запросов
    session_config=session_config,     # Передаем конфигурацию сессии
)

# Функция инициализации базы данных при запуске приложения
async def on_startup() -> None:
    """
    Инициализирует базу данных при старте приложения.
    Создает все таблицы, определенные в метаданных моделей (UUIDBase).
    """
    async with db_config.get_engine().begin() as conn:
        await conn.run_sync(base.UUIDBase.metadata.create_all)

# Инициализация плагина SQLAlchemy для интеграции с Litestar
sqlalchemy_plugin = SQLAlchemyInitPlugin(config=db_config)

# Создание экземпляра приложения Litestar
app = Litestar(
    route_handlers=[UserController],  # Подключаем контроллер пользователей
    on_startup=[on_startup],          # Выполняем функцию инициализации при запуске
    dependencies={"limit_offset": Provide(provide_limit_offset_pagination)},  # Зависимость пагинации
    plugins=[sqlalchemy_plugin, SQLAlchemySerializationPlugin()],  # Подключаем плагины
)

📌 Примечание: В комментариях, подробно указано, что за что отвечает. Если в кратце, то здесь мы обозначили что для подключения к БД будем использовать sqlalchemy, с помощью плагина SQLAlchemyInitPlugin, а с помощью другого плагина SQLAlchemySerializationPlugin будем сериализовать модели SQLAlchemy в JSON. А также добавлена в зависимости пагинация. Сможем использовать её , например для получения списка пользователей.
Остальные модели и контроллеры я добавлю по той же схеме. А вы их можете просто скопировать или попробовать добавить самостоятельно.

Запуск и проверка

🎉 Шаг 1: Запускаем приложение

Запускаем:

make run

Ожидаемый вывод:

❯ make run
uv run granian --interface asgi src/app:app
[INFO] Starting granian (main PID: 485148)
[INFO] Listening at: http://127.0.0.1:8000
[INFO] Spawning worker-1 with pid: 485149
[INFO] Started worker-1
[INFO] Started worker-1 runtime-1

🧪 Шаг 2: Проверяем эндпоинты с curl

• Создание пользователя (POST /users):

curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"first_name": "Мурзик", "last_name": "Котов", "email": "[e-mail]"}'
curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"first_name": "Барсик", "last_name": "Пушистый", "email": "[e-mail]"}'
curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"first_name": "Васька", "last_name": "Лапкин", "email": "[e-mail]"}'

Ожидаемый ответ:

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "first_name": "Мурзик",
  "last_name": "Котов",
  "email": "[e-mail]",
  "created_at": "2023-10-01T12:00:00Z",
  "updated_at": "2023-10-01T12:00:00Z"
}

• Получение списка пользователей (GET /users):

curl -X GET "http://localhost:8000/users?limit=2&offset=0"

Ожидаемый ответ:

{
  "items": [
    {
      "id": "550e8400-e29b-41d4-a716-446655440000",
      "first_name": "Мурзик",
      "last_name": "Котов",
      "email": "[e-mail]",
      "created_at": "2023-10-01T12:00:00Z",
      "updated_at": "2023-10-01T12:00:00Z"
    },
    {
      "id": "550e8400-e29b-41d4-a716-446655440001",
      "first_name": "Барсик",
      "last_name": "Пушистый",
      "email": "[e-mail]",
      "created_at": "2023-10-01T12:05:00Z",
      "updated_at": "2023-10-01T12:05:00Z"
    }
  ],
  "total": 3,
  "limit": 2,
  "offset": 0
}

• Получение пользователя по UUID (GET /users/{user_id:uuid}):

curl -X GET http://localhost:8000/users/123e4567-e89b-12d3-a456-426614174000

Замените 123e4567-e89b-12d3-a456-426614174000, на реальный UUID пользователя, например, из ответа на создание “Мурзика”.
Ожидаемый ответ:

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "first_name": "Мурзик",
  "last_name": "Котов",
  "email": "[e-mail]",
  "created_at": "2023-10-01T12:00:00Z",
  "updated_at": "2023-10-01T12:00:00Z"
}

• Обновление пользователя (PATCH /users/{user_id:uuid})

curl -X PATCH http://localhost:8000/users/550e8400-e29b-41d4-a716-446655440000 \
-H "Content-Type: application/json" \
-d '{"first_name": "Мурзик Updated"}'

Ожидаемый ответ:

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "first_name": "Мурзик Updated",
  "last_name": "Котов",
  "email": "[e-mail]",
  "created_at": "2023-10-01T12:00:00Z",
  "updated_at": "2023-10-01T12:10:00Z"
}

• Удаление пользователя (DELETE /users/{user_id:uuid}):

curl -X DELETE http://localhost:8000/users/123e4567-e89b-12d3-a456-426614174000

Ожидаемый ответ:

Код состояния: 204 No Content
Тело ответа отсутствует.

Итоги

🐾 Поздравляю! База готова. Мы отлично потрудились!

Что мы сделали:

• ✅ Подготовили PostgreSQL в Docker.

• ✅ Настроили тесты с Pytest и TDD.

• ✅ Создали модели данных и миграции с Alembic.

• ✅ Реализовали CRUD-операции с Litestar.

• ✅ Проверили работу API через curl.

Дальше будет интереснее. Нарастим мышцы нашему приложению.

Ссылка на GitHub: https://github.com/pulichkin/litestarcats.git

Читать 1-ую часть.