Взаимодействие с базой данных через SQLAlchemy в FastAPI

Сегодня мы сделаем ещё один шаг вперёд и углубимся в то, как интегрировать SQLAlchemy с FastAPI для выполнения CRUD-операций через API.

Готовы? Наливайте кофе, мы начинаем!

Как только наши модели данных настроены в SQLAlchemy, нам нужно предоставить интерфейс для взаимодействия с этими данными.

И здесь FastAPI показывает себя во всей красе: он позволяет легко создавать маршруты (endpoints) для выполнения CRUD операций.

---

Пример модели данных

Начнём с базовой модели данных, которую мы уже создавали ранее.

Предположим, у нас есть таблица User, содержащая информацию о пользователях.

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Настройка подключения и сессии

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"  # Используем SQLite для примера

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Создаём таблицы
Base.metadata.create_all(bind=engine)

Мы снова используем SQLite для простоты, но вы можете легко заменить это на PostgreSQL или любую другую реляционную базу данных.

Создание маршрутов (endpoints)

Теперь, когда подключение настроено, создадим API-маршруты для выполнения CRUD-операций. Начнём с простого FastAPI-приложения.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

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

Функции для операций

Создадим функции для каждой CRUD-операции: создание, чтение, обновление и удаление.

Создание пользователя

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str

@app.post("/users/", response_model=UserCreate)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Когда пользователь отправляет запрос на создание (POST), мы добавляем новый объект в базу данных, фиксируем изменения с помощью db.commit() и возвращаем созданный объект.

Чтение пользователей

from typing import List

@app.get("/users/", response_model=List[UserCreate])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

Мы используем skip и limit в качестве параметров запроса, чтобы реализовать пагинацию. Это полезно при больших объёмах данных.

Обновление пользователя

@app.put("/users/{user_id}", response_model=UserCreate)
def update_user(user_id: int, updated_user: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    db_user.name = updated_user.name
    db_user.email = updated_user.email
    db.commit()
    db.refresh(db_user)
    return db_user

Мы выполняем поиск пользователя по id, обновляем его данные, фиксируем изменения и возвращаем обновлённый объект.

Удаление пользователя

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"message": "User deleted successfully"}

Здесь мы ищем пользователя, удаляем его и фиксируем изменения в базе данных. Если пользователь не найден, возвращаем ошибку 404.

---

Пример выполнения асинхронных операций

FastAPI поддерживает асинхронные функции, но важно понимать, что пока SQLAlchemy (в своём стандартном виде) работает синхронно.

Для асинхронных операций нам потребуется использовать библиотеку asyncpg, вместе с databases.

Установка необходимых библиотек

pip install databases asyncpg

Асинхронная настройка подключения

Используем библиотеку databases для асинхронного взаимодействия.

from databases import Database

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/testdb"

database = Database(DATABASE_URL)

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

Асинхронные CRUD-операции реализуются практически так же, как и синхронные. Например:

Создание записи

@app.post("/async_users/")
async def create_async_user(user: UserCreate):
    query = "INSERT INTO users(name, email) VALUES (:name, :email)"
    await database.execute(query=query, values={"name": user.name, "email": user.email})
    return {"message": "User created!"}

Чтение записей

@app.get("/async_users/")
async def read_async_users():
    query = "SELECT * FROM users"
    return await database.fetch_all(query=query)

Асинхронные операции особенно полезны при высоких нагрузках или при взаимодействии с другими API.

---

Тестирование CRUD операций через API

Для тестирования мы можем использовать инструменты вроде Postman или curl.

Но автоматическое тестирование — это must-have для любого серьёзного проекта.

Давайте создадим тесты с использованием Pytest и TestClient.

FastAPI предоставляет TestClient для тестирования.

Установка необходимых библиотек

pip install pytest httpx

Пример теста

from fastapi.testclient import TestClient

client = TestClient(app)

def test_create_user():
    response = client.post("/users/", json={"name": "John", "email": "john@example.com"})
    assert response.status_code == 200
    assert response.json()["email"] == "john@example.com"

Этот тест проверяет маршруты POST для создания пользователя. Аналогично можно тестировать чтение, обновление и удаление.

---

Теперь наше приложение полноценно взаимодействует с базой данных через SQLAlchemy и FastAPI.

Мы реализовали основные CRUD операции, изучили асинхронные возможности и автоматическое тестирование.

Это не только полезно для реальных проектов, но и впечатлит любого интервьюера на собеседовании! 🚀