FastAPI + SQLAlchemy 2.0

Собрал на основе статей по FastAPI проект-шаблон для быстрого старта веб-приложения с поддержкой SQLAlchemy 2.0. Здесь добавлено минимально необходимое:

• подготовленный набор пакетов зависимостей;
• миграции на alembic с удобной сортировкой файлов;
• готовый пример работы с БД в pytest;
• базовая модель в БД с полями времени создания и обновления записей в таблицах, команды для быстрой установки виртуального окружения, создания и проведения миграций, проверки стиля кода, запуска тестов, запуска приложения;
• настройка для запуска проекта в докере.

За основу была взята статья на хабре, которая в свою очередь является переводом части с более подробной англоязычной статьи. Могло показаться, что этой статьи достаточно, но мне не хватило некоторых кейсов:

• примера создания базовой модели SQLModel;
• примера работы с БД в тестах с моделями SQLmodel;
• наименования миграций для сортировки файлов в каталоге в порядке их добавления;

Пример создания базовой модели SQLModel.

В моделях наследуемых от BaseModel поля created_at и updated_at будут обновляться автоматически.

from datetime import datetime from typing import Optional from sqlalchemy import BigInteger, Column, text from sqlmodel import Field, SQLModel class BaseModel(SQLModel): id: Optional[int] = Field( default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True)) created_at: datetime = Field( default_factory=datetime.utcnow, nullable=False, sa_column_kwargs=dict(server_default=text('current_timestamp(0)')), ) updated_at: datetime = Field( default_factory=datetime.utcnow, nullable=False, sa_column_kwargs=dict( server_default=text('current_timestamp(0)'), onupdate=text('current_timestamp(0)'), )) class Some(BaseModel, table=True): """Some model.""" other_field: str

Пример работы с БД в тестах с моделями SQLmodel.

Для этого следует задать такой conftest.py с фикстурами:

from typing import Callable, Generator from uuid import uuid4 import pytest_asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import sessionmaker from sqlmodel import SQLModel from core.app import app from core.db_service import async_engine from core.models import Some @pytest_asyncio.fixture async def async_client(async_session: AsyncSession) -> Generator: async with AsyncClient( app=app, base_url='http://localhost/', ) as client: yield client @pytest_asyncio.fixture(scope='function') async def async_session() -> AsyncSession: _async_session = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False) async with _async_session() as _sess: async with async_engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) yield _sess async with async_engine.begin() as conn: await conn.run_sync(SQLModel.metadata.drop_all) await async_engine.dispose() @pytest_asyncio.fixture(scope='function') async def some_factory(async_session: AsyncSession) -> Callable: async def create(**kwargs): params = dict( other_field=f'Test value {uuid4()}', ) params.update(kwargs) some = Some(**params) async_session.add(some) await async_session.commit() return some return create

Здесь core - это модуль самого приложения. Данного кода недостаточно, чтобы это заработало после копирования в ваш проект (нужен еще код функции async_engine), но весь код выкладывать не вижу смысла в статье, потому что его можно увидеть на github.

Наименования миграций для сортировки файлов в каталоге в порядке их добавления.

Для этого в файле конфигурации alembic.ini нужно указать такой параметр:

[alembic] file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s