Интеграция базы данных в FastAPI

FastAPI можно интегрировать с различными базами данных, и наиболее распространённый подход — использовать SQLAlchemy в качестве ORM. В этом разделе показано, как создать API с функциональностью CRUD (создание, чтение, обновление, удаление) с использованием FastAPI + SQLAlchemy.

Установите зависимости

pip install sqlalchemy

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

project/
├── main.py          # FastAPI 应用入口
├── database.py      # 数据库连接配置
├── models.py        # SQLAlchemy 模型
├── schemas.py       # Pydantic 数据模型
└── crud.py          # 数据库操作函数

1. Настройка базы данных

Вdatabase.pyНастройте подключение к базе данных в:

Пример

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# База данных SQLite (подходит для разработки)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# Создание движка базы данных
# connect_args требуется только для SQLite, что обеспечивает многопоточный доступ
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)

# Создание фабрики сессий
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Объявление базового класса
class Base(DeclarativeBase):
pass

Для среды разработки достаточно SQLite. Это файловая база данных, не требующая установки дополнительного сервиса базы данных. Для рабочей среды рекомендуются PostgreSQL или MySQL; вам нужно только изменитьSQLALCHEMY_DATABASE_URLВот и всё.

2. Модель SQLAlchemy

Вmodels.pyОпределите структуру таблицы базы данных в:

Пример

# models.py
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
__tablename__ = "users" # Имя таблицы

id = Column(Integer, primary_key=True, index=True) # Первичный ключ, автоинкремент
email = Column(String, unique=True, index=True) # Уникальное, создать индекс
hashed_password = Column(String) # Хэш пароля
is_active = Column(Boolean, default=True) # Активен

class Item(Base):
__tablename__ = "items"

id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Название, создать индекс
description = Column(String) # Описание
owner_id = Column(Integer, ForeignKey("users.id")) # Внешний ключ ссылается на таблицу users

3. Модели данных Pydantic

Вschemas.pyопределение структур входных/выходных данных API:

Пример

# schemas.py
from pydantic import BaseModel

# ===== Модель пользователя =====
class UserBase(BaseModel):
email: str

class UserCreate(UserBase):
password: str # Пароль обязателен при создании

class UserOut(UserBase):
id: int
is_active: bool

model_config = ConfigDict(from_attributes=True) # Поддерживает объекты ORM

# ===== Модель товара =====
class ItemBase(BaseModel):
title: str
description: str | None = None

class ItemCreate(ItemBase):
pass

class ItemOut(ItemBase):
id: int
owner_id: int

model_config = ConfigDict(from_attributes=True)

Модели Pydantic и модели SQLAlchemy разделены. Pydantic отвечает за валидацию данных на уровне API, а SQLAlchemy — за структуры таблиц на уровне базы данных.from_attributes=TrueРазрешить Pydantic читать данные из объектов SQLAlchemy.

4. Функции операций с базой данных

Вcrud.pyИнкапсулируйте операции с базой данных в:

Пример

# crud.py
from sqlalchemy.orm import Session
import models, schemas

def get_user(db: Session, user_id: int):
"""Получить пользователя по ID"""
return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
"""Получить пользователя по email"""
return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Получить список пользователей"""
return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
"""Создать пользователя"""
fake_hashed_password = user.password + "notreallyhashed" # На практике используйте passlib для хеширования
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Зафиксировать транзакцию
db.refresh(db_user) # Обновить объект, чтобы получить сгенерированный базой данных id
return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Получить список товаров"""
return db.query(models.Item).offset(skip).limit(limit).all()

def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Создать товар"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item

5. Точка входа приложения FastAPI

Вmain.pyИнтегрируйте все компоненты в:

Пример

# main.py
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict

import models, schemas, crud
from database import engine, SessionLocal, Base

# Создать таблицу базы данных
Base.metadata.create_all(bind=engine)

app = FastAPI()

# Зависимость: получить сессию базы данных
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

# ===== Маршруты пользователей =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Проверить, существует ли уже электронная почта
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=“Электронная почта уже зарегистрирована”)
return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users

@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=“Пользователь не существует”)
return db_user

# ===== Маршруты товаров =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
user_id: int,
item: schemas.ItemCreate,
db: Session = Depends(get_db),
):
return crud.create_item(db=db, item=item, user_id=user_id)

@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items

Внедрение зависимостей для сессий базы данных

Основной шаблон — использовать внедрение зависимостей для управления сессиями базы данных:

def get_db():
    db = SessionLocal()   # 创建会话
    try:
        yield db          # 提供会话给路由函数
    finally:
        db.close()        # 请求完成后关闭会话

Каждый запрос получает независимую сессию базы данных, которая автоматически закрывается после завершения запроса, что предотвращает утечки ресурсов.

Строки подключения для разных баз данных

база данных	Пример строки подключения
SQLite	`sqlite:///./sql_app.db`
PostgreSQL	`postgresql://user:password@localhost/dbname`
MySQL	`mysql+pymysql://user:password@localhost/dbname`

Для переключения баз данных достаточно изменить строку подключения; остальной код менять не нужно. Это одно из преимуществ использования ORM.

Итог

Используйте SQLAlchemy в качестве ORM, а FastAPI управляет сессиями базы данных через внедрение зависимостей.
Модель Pydantic (schemas.py) выполняет валидацию данных API, модели SQLAlchemy (models.py) отвечает за структуру таблицы базы данных
CRUD-операции инкапсулированы вcrud.pyв, сохраняя код чистым
Для разработки используйте SQLite; для продакшена переключитесь на PostgreSQL/MySQL, изменив только строку подключения.
from_attributes=TrueРазрешите Pydantic читать данные из ORM-объектов.

Другие расширения

Практический проект

Интеграция базы данных в FastAPI

Установите зависимости

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

1. Настройка базы данных

Пример

2. Модель SQLAlchemy

Пример

3. Модели данных Pydantic

Пример

4. Функции операций с базой данных

Пример

5. Точка входа приложения FastAPI

Пример

Внедрение зависимостей для сессий базы данных

Строки подключения для разных баз данных

Итог