Подключить базу данных к FastAPI: пошаговая инструкция

Что понадобится до начала

Прежде чем писать код, убедитесь, что всё перечисленное ниже на месте. Узнавать о нехватке прав на десятом шаге — потерянное время.

• Python 3.10 или новее. Проверьте: python --version.
• Установленный PostgreSQL 14+ с доступом на создание баз. Подойдёт и локальный сервер, и облачный — главное, чтобы у вас были хост, порт, логин и пароль.
• Утилита pip (идёт с Python) и желательно venv для изоляции зависимостей.
• Любой терминал и текстовый редактор. Примеры проверены на FastAPI 0.115.x и SQLAlchemy 2.0.x с асинхронным драйвером asyncpg.

Вся статья построена вокруг связки FastAPI + SQLAlchemy 2.0 (async) + PostgreSQL. Если вы используете MySQL или SQLite, принцип тот же, но строка подключения и драйвер будут другими — сверьтесь с документацией SQLAlchemy.

Шаг 1. Создайте проект и установите зависимости

Создайте каталог и виртуальное окружение. Все команды ниже — для Linux/macOS; на Windows замените source .venv/bin/activate на .venv\Scripts\activate.

mkdir fastapi_db_demo
cd fastapi_db_demo
python -m venv .venv
source .venv/bin/activate

pip install "fastapi[standard]" sqlalchemy asyncpg

Проверка: выполните pip list. В списке должны быть fastapi, sqlalchemy, asyncpg и uvicorn (устанавливается вместе с fastapi[standard]).

asyncpg — асинхронный драйвер для PostgreSQL, написанный на Cython. Он нужен, чтобы SQLAlchemy работал в связке с асинхронным event-loop FastAPI без блокировки потока.

Шаг 2. Создайте базу данных

Если база ещё не существует, создайте её через psql или любой GUI-клиент:

psql -U postgres -c "CREATE DATABASE demo;"

Проверка: psql -U postgres -d demo -c "SELECT 1;" должен вернуть одну строку с единицей.

Шаг 3. Опишите подключение и сессию

Создайте файл database.py. Здесь живут движок (engine), фабрика сессий и базовый класс моделей.

# database.py
# Версии: SQLAlchemy 2.0.x, asyncpg

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://postgres:YOUR_PASSWORD@localhost:5432/demo"

# echo=True — логирует SQL в консоль; в проде отключите
engine = create_async_engine(DATABASE_URL, echo=True)

# expire_on_commit=False — объекты остаются доступны после commit
async_session = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    """Базовый класс для всех моделей."""
    pass


async def get_session() -> AsyncSession:
    """Зависимость FastAPI: открывает и закрывает сессию."""
    async with async_session() as session:
        yield session

Проверка: файл импортируется без ошибок — python -c "import database".

Обратите внимание: пароль в коде — плохая практика. Для рабочих проектов читайте его из переменной окружения или .env-файла. Здесь оставлено так ради простоты первого запуска.

Шаг 4. Опишите модель

Создайте models.py с одной таблицей — этого достаточно, чтобы проверить связку.

# models.py

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String
from database import Base


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author: Mapped[str] = mapped_column(String(200))
    pages: Mapped[int] = mapped_column(default=0)

Mapped и mapped_column — декларативный стиль SQLAlchemy 2.0. В старых руководствах вы встретите Column(Integer, ...) — он по-прежнему работает, но новый стиль даёт подсказки типов и лучше дружит с IDE.

Шаг 5. Создайте таблицы

Добавьте в проект файл create_tables.py. Он запускается один раз, чтобы применить схему к базе.

# create_tables.py

import asyncio
from database import engine, Base
from models import Book  # noqa: F401 — импорт нужен, чтобы Base "увидел" модель


async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    await engine.dispose()


asyncio.run(init_db())

Запустите: python create_tables.py.

Проверка: подключитесь к базе и убедитесь, что таблица создана:

psql -U postgres -d demo -c "\dt"

В списке должна появиться таблица books.

Почему не Alembic

Для первого запуска Base.metadata.create_all проще всего. Но в проекте, где схема меняется, вам понадобится Alembic — инструмент миграций для SQLAlchemy. Это отдельная тема; здесь мы её не разворачиваем.

Шаг 6. Напишите эндпоинты

Создайте main.py — точку входа приложения.

# main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from pydantic import BaseModel

from database import get_session
from models import Book

app = FastAPI()


# --- Pydantic-схемы ---

class BookCreate(BaseModel):
    title: str
    author: str
    pages: int = 0


class BookRead(BaseModel):
    id: int
    title: str
    author: str
    pages: int

    model_config = {"from_attributes": True}


# --- Эндпоинты ---

@app.post("/books", response_model=BookRead)
async def create_book(data: BookCreate, session: AsyncSession = Depends(get_session)):
    book = Book(**data.model_dump())
    session.add(book)
    await session.commit()
    await session.refresh(book)
    return book


@app.get("/books", response_model=list[BookRead])
async def list_books(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(Book))
    return result.scalars().all()


@app.get("/books/{book_id}", response_model=BookRead)
async def get_book(book_id: int, session: AsyncSession = Depends(get_session)):
    book = await session.get(Book, book_id)
    if not book:
        raise HTTPException(status_code=404, detail="Книга не найдена")
    return book

Запуск:

uvicorn main:app --reload

Проверка: откройте http://127.0.0.1:8000/docs. Swagger UI покажет три эндпоинта. Попробуйте создать книгу через форму POST и затем получить список через GET.

Структура проекта

После всех шагов каталог выглядит так:

Для учебного проекта этого достаточно. В растущем приложении модели, схемы и роутеры выносят в отдельные пакеты, но принцип подключения базы остаётся тем же.

Если не получилось

1. ModuleNotFoundError: No module named 'asyncpg' — вы не активировали виртуальное окружение или устанавливали пакеты в другое. Выполните source .venv/bin/activate и повторите pip install asyncpg.
2. sqlalchemy.exc.OperationalError: ... password authentication failed — неверный логин или пароль в DATABASE_URL. Проверьте, что пользователь существует и пароль правильный.
3. connection refused на порту 5432 — PostgreSQL не запущен. На Linux: sudo systemctl start postgresql. На macOS через Homebrew: brew services start postgresql@16.
4. RuntimeError: Task got Future attached to a different loop — скорее всего, вы смешиваете синхронный и асинхронный код. Убедитесь, что используете create_async_engine, а не create_engine, и что get_session — асинхронный генератор.
5. Таблица не создалась, но ошибок нет — вы не импортировали модель до вызова create_all. Без импорта Base.metadata пуст и создавать нечего. Добавьте from models import Book в create_tables.py.

Что делать дальше

Рабочая связка поднята — дальше стоит закрыть три вещи, прежде чем проект дорастёт до прода:

• Миграции. Подключите Alembic, чтобы изменения схемы применялись инкрементально, а не через пересоздание таблиц.
• Конфигурация. Вынесите DATABASE_URL в переменные окружения. Библиотека pydantic-settings позволяет читать .env-файл и валидировать настройки при старте.
• Пул соединений. У create_async_engine есть параметры pool_size и max_overflow. Под нагрузкой значения по умолчанию (5 и 10) могут не хватить — следите за метрикой ожидания свободного соединения.