Какие БД можно использовать в FastAPI?

Question

claude-haiku-4.5 · Accepted Answer

## Использование баз данных в FastAPI

FastAPI — это фреймворк для создания API, он агностичен к БД. Ты можешь использовать любую БД, которая имеет Python драйвер. Выбор зависит от требований проекта.

## Реляционные БД (SQL)

### PostgreSQL (рекомендуется)

```python
# requirements.txt
fastapi==0.104.1
sqlalchemy==2.0.23
psycopg[binary]==3.17
alembic==1.12.1

# models.py
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime, timezone

# Подключение к PostgreSQL
DATABASE_URL = "postgresql://user:password@localhost:5432/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), default=datetime.now(timezone.utc))

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from typing import List

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

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

@app.get("/users")
async def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()
```

### MySQL / MariaDB

```python
# requirements.txt
mysqlclient==2.2.0
# или
mysql-connector-python==8.2.0
# или (рекомендуется)
pymysql==1.1.0

# Подключение
DATABASE_URL = "mysql+pymysql://user:password@localhost:3306/dbname"
engine = create_engine(DATABASE_URL)
```

### SQLite (для разработки)

```python
# Идеально для локальной разработки и тестирования
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # Нужно для SQLite
)
```

## NoSQL БД

### MongoDB

```python
# requirements.txt
motorengine==2.2.4  # Асинхронный драйвер
pydantic==2.0.0

# models.py
from pydantic import BaseModel
from motorengine import Document, StringField, IntField, DateTimeField
from datetime import datetime, timezone

class User(Document):
    email = StringField(required=True)
    age = IntField()
    created_at = DateTimeField(default=datetime.now(timezone.utc))

# main.py
from fastapi import FastAPI
from motorengine import connect

app = FastAPI()

@app.on_event("startup")
async def startup():
    await connect("mongodb://localhost:27017/dbname")

@app.post("/users")
async def create_user(email: str):
    user = User(email=email)
    await user.save()
    return user

@app.get("/users")
async def list_users():
    return await User.objects.all()
```

### Redis (кэш и очереди)

```python
# requirements.txt
redis==5.0.0

# main.py
from fastapi import FastAPI
import redis.asyncio as redis
import json

app = FastAPI()
redis_client = None

@app.on_event("startup")
async def startup():
    global redis_client
    redis_client = await redis.from_url("redis://localhost")

@app.post("/cache")
async def set_cache(key: str, value: str):
    await redis_client.set(key, value, ex=3600)  # 1 час
    return {"status": "ok"}

@app.get("/cache/{key}")
async def get_cache(key: str):
    value = await redis_client.get(key)
    return {"value": value.decode() if value else None}
```

## Облачные БД

### Firebase (Google)

```python
# requirements.txt
firebase-admin==6.3.0

# main.py
from fastapi import FastAPI
import firebase_admin
from firebase_admin import credentials, db

cred = credentials.Certificate("serviceAccountKey.json")
firebase_admin.initialize_app(cred, {
    "databaseURL": "https://project.firebaseio.com"
})

app = FastAPI()

@app.post("/users")
async def create_user(email: str):
    ref = db.reference("users")
    new_ref = ref.push({"email": email})
    return {"id": new_ref.key}

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    ref = db.reference(f"users/{user_id}")
    return ref.get()
```

### DynamoDB (AWS)

```python
# requirements.txt
boto3==1.34.0

# main.py
from fastapi import FastAPI
import boto3
from datetime import datetime, timezone

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('users')

app = FastAPI()

@app.post("/users")
async def create_user(email: str):
    table.put_item(Item={
        "email": email,
        "created_at": datetime.now(timezone.utc).isoformat()
    })
    return {"status": "created"}

@app.get("/users/{email}")
async def get_user(email: str):
    response = table.get_item(Key={"email": email})
    return response.get("Item")
```

## Graph БД

### Neo4j

```python
# requirements.txt
neo4j==5.14.0

# main.py
from fastapi import FastAPI
from neo4j import AsyncGraphDatabase

app = FastAPI()
driver = None

@app.on_event("startup")
async def startup():
    global driver
    driver = AsyncGraphDatabase.driver(
        "neo4j://localhost:7687",
        auth=("neo4j", "password")
    )

@app.post("/relationships")
async def create_relationship(user1: str, user2: str):
    async with driver.session() as session:
        await session.run(
            "MATCH (a:User {name: $user1}), (b:User {name: $user2}) "
            "CREATE (a)-[:KNOWS]->(b)",
            user1=user1,
            user2=user2
        )
    return {"status": "connected"}

@app.on_event("shutdown")
async def shutdown():
    await driver.close()
```

## Выбор БД: сравнение

```python
# PostgreSQL — ЛУЧШИЙ выбор для большинства случаев
# ✓ Мощная, надёжная
# ✓ ACID транзакции
# ✓ Полнотекстовый поиск
# ✓ JSON типы
# ✓ Расширения (PostGIS для геоданных)

# MySQL — хороший выбор для веб-приложений
# ✓ Быстрая для чтения
# ✓ Простая в настройке
# ✗ Менее мощная чем PostgreSQL

# SQLite — только для разработки и малых проектов
# ✓ Не требует сервера
# ✓ Идеально для тестов
# ✗ Не подходит для production с высокой нагрузкой

# MongoDB — для неструктурированных данных
# ✓ Гибкая схема
# ✓ Хороша для документов
# ✗ Нет ACID транзакций (в старых версиях)
# ✗ Требует больше памяти

# Redis — только как кэш/очередь
# ✓ Очень быстрый
# ✗ Хранит в памяти
# ✗ Нет персистентности по умолчанию
```

## Лучшие практики

```python
# ✓ ПРАВИЛЬНО
from sqlalchemy.orm import Session
from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
async def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

# ✓ Используй connection pooling
engine = create_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=0,
    pool_pre_ping=True
)

# ✓ Используй миграции (Alembic)
# alembic init alembic
# alembic revision --autogenerate -m "Add users table"
# alembic upgrade head

# ✗ НЕПРАВИЛЬНО
# Создавать новое подключение для каждого запроса
db = SessionLocal()  # ✗ ПЛОХО!

# Использовать raw SQL везде
db.execute("SELECT * FROM users")  # ✗ Уязвимо для SQL injection
```

## Асинхронные драйверы

```python
# requirements.txt
sqlalchemy==2.0.23
asyncpg==0.28.0  # Асинхронный драйвер для PostgreSQL

# DATABASE_URL для асинхронной работы
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

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

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

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

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    from sqlalchemy import select
    result = await db.execute(select(User))
    return result.scalars().all()
```

## Итого

Выбор БД для FastAPI:
- **PostgreSQL** — рекомендуется для production
- **MySQL** — альтернатива для веб-приложений
- **SQLite** — только для разработки
- **MongoDB** — для неструктурированных данных
- **Redis** — как кэш и очередь
- Используй асинхронные драйверы (asyncpg) для лучшей производительности
- Всегда используй фикстуры зависимостей для управления БД сессиями

Какие БД можно использовать в FastAPI?

Комментарии (1)

Использование баз данных в FastAPI

Реляционные БД (SQL)

PostgreSQL (рекомендуется)

MySQL / MariaDB

SQLite (для разработки)

NoSQL БД

MongoDB

Redis (кэш и очереди)

Облачные БД

Firebase (Google)

DynamoDB (AWS)

Graph БД

Neo4j

Выбор БД: сравнение

Лучшие практики

Асинхронные драйверы

Итого