Nun erweitern wir die kleine FastAPI-REST-API so, dass sie nicht nur mit einer Liste im Speicher arbeitet, sondern mit einer echten SQLite-Datenbank. Dazu nutzen wir SQLAlchemy als ORM (Object Relational Mapper).
Installation
pip install fastapi uvicorn sqlalchemy
Beispiel: REST-API mit SQLite & SQLAlchemy
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
# FastAPI App
app = FastAPI()
# SQLite DB einrichten
DATABASE_URL = "sqlite:///./users.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# DB-Modell
class UserDB(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False)
email = Column(String, unique=True, index=True, nullable=False)
# Tabellen erstellen
Base.metadata.create_all(bind=engine)
# Pydantic Modell (für API)
class User(BaseModel):
id: int
name: str
email: str
class Config:
orm_mode = True
# Dependency für DB-Session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 1) Alle User abrufen
@app.get("/users", response_model=list[User])
def get_users(db: Session = Depends(get_db)):
return db.query(UserDB).all()
# 2) User nach ID abrufen
@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(UserDB).filter(UserDB.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
# 3) Neuen User anlegen
@app.post("/users", response_model=User, status_code=201)
def create_user(user: User, db: Session = Depends(get_db)):
db_user = UserDB(id=user.id, name=user.name, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
# 4) User aktualisieren
@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, updated_user: User, db: Session = Depends(get_db)):
user = db.query(UserDB).filter(UserDB.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
user.name = updated_user.name
user.email = updated_user.email
db.commit()
db.refresh(user)
return user
# 5) User löschen
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(UserDB).filter(UserDB.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
db.delete(user)
db.commit()
return
Server starten
uvicorn main:app --reload
Jetzt wird automatisch eine SQLite-Datenbank users.db im Projektordner angelegt, und die API arbeitet mit echten Tabellen.
Vorteile gegenüber der Listen-Version
-
Persistenz → Daten bleiben nach Server-Neustart erhalten.
-
Echte Datenbankfeatures → Indizes, Constraints, Joins usw.
-
Skalierbarkeit → später einfach auf PostgreSQL, MySQL oder eine Cloud-DB wechseln.
