О чем важно помнить, когда реализуешь связь между FastAPI и PostgreSQL

Управление подключениями к БД

При работе с FastAPI и PostgreSQL критически важно правильно управлять пулом соединений. Используй asyncpg или SQLAlchemy async вместо синхронных драйверов. Каждое соединение — это ресурс, и его неправильное использование приводит к deadlock'ам и истощению пула.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    pool_size=20,
    max_overflow=10,
    echo=False,
    pool_pre_ping=True,  # Проверка соединения перед использованием
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    future=True
)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

Транзакции и изоляция

Помни о уровнях изоляции в PostgreSQL (READ UNCOMMITTED, READ COMMITTED, REPEATABLE READ, SERIALIZABLE). По умолчанию используется READ COMMITTED, но для критичных операций нужна SERIALIZABLE. Явно управляй транзакциями:

@app.post("/orders")
async def create_order(order_data: OrderSchema, session: AsyncSession = Depends(get_db)):
    async with session.begin():
        result = await session.execute(insert(Order).values(**order_data.dict()))
        await session.flush()
        return {"id": result.lastrowid}

Обработка ошибок БД

Никогда не логируй ошибки БД на верхних слоях — они должны обрабатываться там, где возникают. Конвертируй специфичные ошибки PostgreSQL в доменные исключения:

from sqlalchemy.exc import IntegrityError, OperationalError

try:
    async with session.begin():
        await session.execute(insert(User).values(email=email))
except IntegrityError as e:
    raise UserAlreadyExistsError(f"Email {email} already in use")
except OperationalError as e:
    raise DatabaseConnectionError("Cannot connect to database")

Подготовленные запросы

Всегда используй параметризованные запросы для защиты от SQL-инъекций:

from sqlalchemy import text

result = await session.execute(
    text("SELECT * FROM users WHERE id = :user_id"),
    {"user_id": user_id}
)

Производительность

Оптимизируй N+1 проблему через eager loading:

from sqlalchemy.orm import selectinload
from sqlalchemy import select

query = select(User).options(selectinload(User.posts))
users = await session.execute(query)

Таймауты

Устанавливай таймауты на операции:

engine = create_async_engine(
    "postgresql+asyncpg://...",
    connect_args={
        "timeout": 10,
        "command_timeout": 60,
    }
)

Эти принципы обеспечивают надежность и безопасность.