7-10. FastAPI + SQLAlchemy による RESTful CRUD API

FastAPI、Pydantic、SQLAlchemyを組み合わせ、SQLiteデータベースを操作する安全で高速なバックエンドAPIを構築します。

Python

from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker, Session

# データベース設定（第4章）
DATABASE_URL = "sqlite:///./app_handson.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class ItemModel(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    price = Column(Integer, nullable=False)

Base.metadata.create_all(bind=engine)

# Pydantic スキーマ
class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    price: int = Field(..., ge=0)

class ItemResponse(ItemCreate):
    id: int
    class Config:
        from_attributes = True

# FastAPIアプリ定義
app = FastAPI(title="実践CRUD API")

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

@app.post("/items/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = ItemModel(name=item.name, price=item.price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item