Подключение PostgreSQL к FastAPI: методы и примеры кода

Дополнительные примеры и расширенное использование

Пример 1: Сложный запрос с JOIN и пагинацией (асинхронный SQLAlchemy)

Предположим, есть таблицы User и Post. Нужно получить посты пользователя с постраничным выводом.

# models.py
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    title = Column(String)
    content = Column(String)
    user = relationship("User", back_populates="posts")

User.posts = relationship("Post", back_populates="user")

# endpoint
@app.get("/users/{user_id}/posts/")
async def get_user_posts(user_id: int, page: int = 1, per_page: int = 10, session: AsyncSession = Depends(get_session)):
    offset = (page - 1) * per_page
    stmt = (
        select(Post)
        .where(Post.user_id == user_id)
        .offset(offset)
        .limit(per_page)
    )
    result = await session.execute(stmt)
    posts = result.scalars().all()
    return {"page": page, "per_page": per_page, "posts": posts}

Результат для страницы 1:

{
  "page": 1,
  "per_page": 10,
  "posts": [
    {"id": 1, "user_id": 1, "title": "Post 1", "content": "Content 1"},
    ...
  ]
}

Такой запрос эффективно использует LIMIT и OFFSET для пагинации.

Пример 2: Миграции с Alembic для асинхронного движка

Установка: pip install alembic. Инициализация: alembic init alembic. Редактируем env.py для асинхронной работы.

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context
from database import Base  # импорт вашего Base
import asyncio

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    connectable = create_async_engine(
        config.get_main_option("sqlalchemy.url"),
        poolclass=pool.NullPool
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

def run_migrations_online():
    asyncio.run(run_async_migrations())

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

В alembic.ini укажите строку подключения: sqlalchemy.url = postgresql+asyncpg://user:pass@localhost/dbname.

Создайте миграцию: alembic revision --autogenerate -m "initial". Примените: alembic upgrade head.

Результат:

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 1234abcd, initial

Пример 3: Транзакции с асинхронным SQLAlchemy

Использование контекстного менеджера сессии гарантирует атомарность операций.

@app.post("/transfer/")
async def transfer_funds(from_id: int, to_id: int, amount: float, session: AsyncSession = Depends(get_session)):
    async with session.begin():
        from_account = await session.get(Account, from_id)
        to_account = await session.get(Account, to_id)
        if not from_account or not to_account:
            raise HTTPException(404, "Account not found")
        if from_account.balance < amount:
            raise HTTPException(400, "Insufficient funds")
        from_account.balance -= amount
        to_account.balance += amount
    return {"status": "ok"}

Результат при успешном переводе:

{"status": "ok"}

Пример 4: Тестирование с использованием mock и in-memory базы

Для unit-тестирования можно заменить реальную базу на SQLite в памяти.

# test_main.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from database import Base, get_session
from main import app
from httpx import AsyncClient

@pytest.fixture
async def test_db():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async def override_get_session():
        async with async_session() as session:
            yield session

    app.dependency_overrides[get_session] = override_get_session
    yield

@pytest.mark.asyncio
async def test_create_item(test_db):
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/items/", params={"name": "test", "description": "desc"})
    assert response.status_code == 200
    assert response.json()["name"] == "test"

Результат выполнения теста:

1 passed in 0.45s

Пример 5: Использование asyncpg с пулом напрямую для массовой вставки

Массовая вставка с помощью executemany значительно ускоряет вставку большого количества строк.

async def bulk_insert_items(items: list[dict]):
    async with app.state.db_pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO items(name, description) VALUES ($1, $2)",
            [(item["name"], item["description"]) for item in items]
        )

Результат:

Вставлено 1000 записей за 0.2 секунды.

Пример 6: Отношения многие-ко-многим с SQLAlchemy

Таблицы Student и Course с ассоциативной таблицей.

# models.py
association_table = Table("student_courses", Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id")),
    Column("course_id", Integer, ForeignKey("courses.id"))
)

class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    courses = relationship("Course", secondary=association_table, back_populates="students")

class Course(Base):
    __tablename__ = "courses"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    students = relationship("Student", secondary=association_table, back_populates="courses")

Запрос студентов с курсами:

from sqlalchemy.orm import selectinload

@app.get("/students/{student_id}/")
async def get_student_with_courses(student_id: int, session: AsyncSession = Depends(get_session)):
    stmt = select(Student).options(selectinload(Student.courses)).where(Student.id == student_id)
    result = await session.execute(stmt)
    student = result.scalar_one_or_none()
    if not student:
        raise HTTPException(404)
    return {"id": student.id, "name": student.name, "courses": [{"id": c.id, "title": c.title} for c in student.courses]}

Результат:

{
  "id": 1,
  "name": "Alice",
  "courses": [
    {"id": 1, "title": "Mathematics"},
    {"id": 2, "title": "Physics"}
  ]
}