SavePearlHarbor

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

—

от автора

admin

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

  • Поднять 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="vasyamurzikov@whiskers.com",  # Почта котика     )     # Создаем объект пользователя из данных     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 == "vasyamurzikov@whiskers.com")     )     # Получаем пользователя из результата запроса     fetched_user = result.scalar_one()     # Проверяем, что имя и email пользователя соответствуют ожидаемым значениям     assert fetched_user.first_name == "Васька"     assert fetched_user.email == "vasyamurzikov@whiskers.com"

🚀 Шаг 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": "murzik@example.com"}' curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"first_name": "Барсик", "last_name": "Пушистый", "email": "barsik@example.com"}' curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"first_name": "Васька", "last_name": "Лапкин", "email": "vaska@example.com"}'

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

{   "id": "550e8400-e29b-41d4-a716-446655440000",   "first_name": "Мурзик",   "last_name": "Котов",   "email": "murzik@example.com",   "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": "murzik@example.com",       "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": "barsik@example.com",       "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": "murzik@example.com",   "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": "murzik@example.com",   "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


ссылка на оригинал статьи https://habr.com/ru/articles/889022/

Комментарии

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *