FastAPI-Datenbankintegration
FastAPI kann mit verschiedenen Datenbanken integriert werden, und der gebräuchlichste Ansatz ist die Verwendung von SQLAlchemy als ORM. In diesem Abschnitt wird erläutert, wie man mit FastAPI + SQLAlchemy eine API mit CRUD-Funktionalität (Create, Read, Update, Delete) erstellt.
Abhängigkeiten installieren
pip install sqlalchemy
Projektstruktur
project/ ├── main.py # FastAPI 应用入口 ├── database.py # 数据库连接配置 ├── models.py # SQLAlchemy 模型 ├── schemas.py # Pydantic 数据模型 └── crud.py # 数据库操作函数
1. Datenbankkonfiguration
Indatabase.pyKonfigurieren Sie die Datenbankverbindung in:
Beispiel
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
# SQLite-Datenbank (für die Entwicklung geeignet)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Datenbank-Engine erstellen
# connect_args wird nur für SQLite benötigt und ermöglicht mehrthreadigen Zugriff
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# Session-Factory erstellen
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Basisklasse deklarieren
class Base(DeclarativeBase):
pass
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
# SQLite-Datenbank (für die Entwicklung geeignet)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Datenbank-Engine erstellen
# connect_args wird nur für SQLite benötigt und ermöglicht mehrthreadigen Zugriff
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# Session-Factory erstellen
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Basisklasse deklarieren
class Base(DeclarativeBase):
pass
Für die Entwicklungsumgebung ist SQLite ausreichend. Es ist eine dateibasierte Datenbank, die keine zusätzliche Installation eines Datenbankdienstes erfordert. Für die Produktionsumgebung werden PostgreSQL oder MySQL empfohlen; Sie müssen nur die
SQLALCHEMY_DATABASE_URLDas war's.
2. SQLAlchemy-Modell
Inmodels.pyDefiniere die Datenbanktabellenstruktur in:
Beispiel
# models.py
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # Tabellenname
id = Column(Integer, primary_key=True, index=True) # Primärschlüssel, Auto-Inkrement
email = Column(String, unique=True, index=True) # Eindeutig, Index erstellen
hashed_password = Column(String) # Passwort-Hash
is_active = Column(Boolean, default=True) # Ist aktiv
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Titel, Index erstellen
description = Column(String) # Beschreibung
owner_id = Column(Integer, ForeignKey("users.id")) # Fremdschlüssel verweist auf die users-Tabelle
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # Tabellenname
id = Column(Integer, primary_key=True, index=True) # Primärschlüssel, Auto-Inkrement
email = Column(String, unique=True, index=True) # Eindeutig, Index erstellen
hashed_password = Column(String) # Passwort-Hash
is_active = Column(Boolean, default=True) # Ist aktiv
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Titel, Index erstellen
description = Column(String) # Beschreibung
owner_id = Column(Integer, ForeignKey("users.id")) # Fremdschlüssel verweist auf die users-Tabelle
3. Pydantic-Datenmodelle
Inschemas.pyDefinition der Eingabe-/Ausgabedatenstrukturen der API:
Beispiel
# schemas.py
from pydantic import BaseModel
# ===== Benutzermodell =====
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str # Passwort erforderlich beim Erstellen
class UserOut(UserBase):
id: int
is_active: bool
model_config = ConfigDict(from_attributes=True) # Unterstützt ORM-Objekte
# ===== Produktmodell =====
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)
from pydantic import BaseModel
# ===== Benutzermodell =====
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str # Passwort erforderlich beim Erstellen
class UserOut(UserBase):
id: int
is_active: bool
model_config = ConfigDict(from_attributes=True) # Unterstützt ORM-Objekte
# ===== Produktmodell =====
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-Modelle und SQLAlchemy-Modelle sind getrennt. Pydantic ist für die Datenvalidierung auf der API-Ebene verantwortlich, während SQLAlchemy für die Tabellenstrukturen auf der Datenbankebene verantwortlich ist.
from_attributes=TruePydantic in die Lage versetzen, Daten aus SQLAlchemy-Objekten zu lesen.
4. Funktionen für Datenbankoperationen
Incrud.pyDatenbankoperationen kapseln in:
Beispiel
# crud.py
from sqlalchemy.orm import Session
import models, schemas
def get_user(db: Session, user_id: int):
"""Benutzer nach ID abrufen"""
return db.query(models.User).filter(models.User.user_id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""Benutzer per E-Mail abrufen"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Benutzerliste abrufen"""
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
"""Benutzer erstellen"""
fake_hashed_password = user.password + "notreallyhashed" # In der Praxis passlib zum Hashing verwenden
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Transaktion festschreiben
db.refresh(db_user) # Objekt aktualisieren, um die datenbankgenerierte ID zu erhalten
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Produktliste abrufen"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Produkt erstellen"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
from sqlalchemy.orm import Session
import models, schemas
def get_user(db: Session, user_id: int):
"""Benutzer nach ID abrufen"""
return db.query(models.User).filter(models.User.user_id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""Benutzer per E-Mail abrufen"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Benutzerliste abrufen"""
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
"""Benutzer erstellen"""
fake_hashed_password = user.password + "notreallyhashed" # In der Praxis passlib zum Hashing verwenden
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Transaktion festschreiben
db.refresh(db_user) # Objekt aktualisieren, um die datenbankgenerierte ID zu erhalten
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Produktliste abrufen"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Produkt erstellen"""
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-Anwendungseinstiegspunkt
Inmain.pyIntegrieren Sie alle Komponenten in:
Beispiel
# 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
# Datenbanktabelle erstellen
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Abhängigkeit: Datenbanksitzung abrufen
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ===== Benutzerrouten =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Prüfen, ob die E-Mail bereits existiert
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=“E-Mail bereits registriert”)
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=“Benutzer existiert nicht”)
return db_user
# ===== Produkt-Routen =====
@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
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
# Datenbanktabelle erstellen
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Abhängigkeit: Datenbanksitzung abrufen
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ===== Benutzerrouten =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Prüfen, ob die E-Mail bereits existiert
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=“E-Mail bereits registriert”)
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=“Benutzer existiert nicht”)
return db_user
# ===== Produkt-Routen =====
@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
Dependency Injection für Datenbank-Sessions
Das Kernmuster besteht darin, Dependency Injection zu verwenden, um Datenbank-Sessions zu verwalten:
def get_db():
db = SessionLocal() # 创建会话
try:
yield db # 提供会话给路由函数
finally:
db.close() # 请求完成后关闭会话
Jede Anfrage erhält eine unabhängige Datenbank-Session, die nach Abschluss der Anfrage automatisch geschlossen wird, wodurch Ressourcenlecks vermieden werden.
Verbindungszeichenfolgen für verschiedene Datenbanken
| Datenbank | Beispiel für eine Verbindungszeichenfolge |
|---|---|
| SQLite | sqlite:///./sql_app.db |
| PostgreSQL | postgresql://user:password@localhost/dbname |
| MySQL | mysql+pymysql://user:password@localhost/dbname |
Der Wechsel der Datenbank erfordert nur das Ändern der Verbindungszeichenfolge; der restliche Code muss nicht geändert werden. Dies ist einer der Vorteile der Verwendung eines ORM.
Zusammenfassung
- Verwenden Sie SQLAlchemy als ORM, und FastAPI verwaltet Datenbank-Sessions durch Dependency Injection.
- Pydantic-Modell (
schemas.py) übernimmt die Validierung der API-Daten, SQLAlchemy-Modelle (models.py) ist für die Struktur der Datenbanktabelle verantwortlich - CRUD-Operationen sind gekapselt in
crud.pyin, wodurch der Code sauber bleibt - Für die Entwicklung verwenden Sie SQLite; für die Produktion wechseln Sie zu PostgreSQL/MySQL, indem Sie nur die Verbindungszeichenfolge ändern.
from_attributes=TrueErmöglichen Sie Pydantic, Daten aus ORM-Objekten zu lesen.