FastAPIを使ってみる

FastAPI, poetry, uvicorn, alembicを使ってみます。

• FastAPIはWEBフレームワークですね。API作ると、SwaggerUIを勝手に作ってくれるらしいです。
• poetryはnpmみたいなやつでライブラリを簡単に追加して依存関係を管理してくれる新しい便利なやつらしいです。
• uvicornはWEBブラウザなんですかね？
• alembicはマイグレーションツールなんですかね？

FastAPI

mkdir fastapi-1
cd fastapi-1
git init
poetry init
poetry add fastapi
poetry add uvicorn[standard]

main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

uvicorn main:app --reload

これで起動して、SwaggerUIまで表示できる。便利そう。

FastAPIでDB接続、マイグレーション、ORMなど

今自分のローカル環境には、PostgreSQLが入っていて既に動いています。

マイグレーション

poetry add sqlalchemy alembic psycopg2-binary
alembic init alembic
alembic revision -m "create users table"

alembicはsqlalchemyと連動するマイグレーションツールとのことです。 上記でalembicの初期化と最初のマイグレーションファイルを作成しました。 このやり方だと、マイグレーションファイルは手動で作成する必要があります。

"""create users table

Revision ID: de05c422a1ac
Revises:
Create Date: 2024-12-13 10:48:01.319333

"""

from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = "de05c422a1ac"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("name", sa.String(50), nullable=False),
        sa.Column("email", sa.String(255), nullable=False),
        sa.Column("password", sa.String(255), nullable=False),
        sa.Column("created_at", sa.DateTime, nullable=False),
        sa.Column("updated_at", sa.DateTime, nullable=False),
    )


def downgrade() -> None:
    op.drop_table("users")

マイグレーション実行時のDB接続については、alembic.ini の sqlalchemy.url を環境に合わせて更新します。

sqlalchemy.url = postgresql://postgres:password@localhost:5432/hoge

これでマイグレーション実行可能になりましたので、下記で実行します。

alembic upgrade head

色々なコマンドがここに書いてあります。

FastAPIでDB接続・ORM

touch .env
mkdir app
mv main.py app
cd app
touch config.py
touch database.py
touch models.py
touch schemas.py

configy.pyで.envの環境変数を読み込み、他のファイルではconfigy.pyを読み込んで使います。 database.pyでDBと接続します。 models.pyにテーブルの定義を書きます。 schemas.pyにリクエストやレスポンス時の型を定義します。 みたいな感じっぽいです。

config.py

from dotenv import load_dotenv
import os

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL")

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from .config import DATABASE_URL

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# DBセッション依存性
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

models.py

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from .database import Base


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    email = Column(String, unique=True, index=True)
    password = Column(String, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)

schemas.py

from pydantic import BaseModel


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


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int

    class Config:
        from_attributes = True

main.py

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from . import models, schemas
from .database import engine, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}


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


@app.get("/users/", response_model=list[schemas.User])
def read_users(db: Session = Depends(get_db)):
    return db.query(models.User).all()

main.pyにある、models.Base.metadata.create_all(bind=engine) は、存在しないテーブルやカラムを勝手に作成してくれるそうです。 ですので、1にで開発中は、テーブル全部削除して、 uvicorn app.main:app --reload をすればテーブルが最新状態で作成されます。 これは便利かも。まあでも、毎回テーブル削除するスクリプトとかは必要かも。

claude先生によるとdb.commit()した時点で、idをdb_userオブジェクトに入れてくれるそうです。 ですので、commit()した時点でidはdb_user.idで取得可能になるそうじゃ。 db.refresh(db_user)はselect文でこのユーザを改めて取得しているようなもんだそうじゃ。 例えば、created_atとかは、DBが勝手に保存するので、そのような値が必要な場合に使えるそうじゃ。 逆に言うと、idだけ返せばよいなら、refreshは不要なんだそうじゃ。 まあ試してないど、多分本当のことじゃろう。便利じゃな。

SQLModelを使って簡略化

下記を見ると、SQLAlchemyではなく、SQLModelというのを使っているようです。

fastapi.tiangolo.com

SQLModel は FastAPI の作者である Tiangolo が作った、以下の2つのライブラリを組み合わせたものです: - SQLAlchemy (ORM) - Pydantic (データバリデーション) つまり、SQLModel は SQLAlchemy の機能を全て継承しつつ、FastAPI との相性を良くするために Pydantic との統合を強化したライブラリなんです。

だそうです。つまり、SQLAlchemyを使っていて問題はないのですが、よりスッキリかけるやつということのようです。 これを使って書き直してみましょう。

poetry add sqlmodel

models.py

from datetime import datetime, UTC
from typing import Optional
from sqlmodel import SQLModel, Field, create_engine, Session
from .config import DATABASE_URL

# データベース設定
engine = create_engine(DATABASE_URL)


# モデル定義（schemas.pyとmodels.pyが統合される）
class UserBase(SQLModel):
    name: str
    email: str = Field(unique=True, index=True)


class User(UserBase, table=True):
    __tablename__ = "users"
    id: Optional[int] = Field(default=None, primary_key=True)
    password: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        sa_column_kwargs={"onupdate": lambda: datetime.now(UTC)},
    )


class UserCreate(UserBase):
    password: str


# DBセッション依存性
def get_db():
    with Session(engine) as session:
        yield session


# アプリ起動時にテーブルを作成
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from sqlmodel import Session, select
from . import models

@asynccontextmanager
async def lifespan(app: FastAPI):
    models.create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}


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


@app.get("/users/", response_model=list[models.User])
def read_users(db: Session = Depends(models.get_db)):
    return db.exec(select(models.User)).all()

こんな感じでできました。

バリデーションチェック

バリデーションルールもmodelsに書けるようです。

models.py

from datetime import datetime, UTC
from typing import Optional
from sqlmodel import SQLModel, Field, create_engine, Session
from pydantic import EmailStr
from .config import DATABASE_URL

# データベース設定
engine = create_engine(DATABASE_URL)


# モデル定義（schemas.pyとmodels.pyが統合される）
class UserBase(SQLModel):
    name: str = Field(
        min_length=1,
        max_length=50,
        description="ユーザー名は1-50文字で入力してください"
    )
    email: EmailStr = Field(
        unique=True,
        index=True,
        description="有効なメールアドレスを入力してください"
    )


class User(UserBase, table=True):
    __tablename__ = "users"
    id: Optional[int] = Field(default=None, primary_key=True)
    password: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        sa_column_kwargs={"onupdate": lambda: datetime.now(UTC)},
    )


class UserCreate(UserBase):
    password: str = Field(
        min_length=8,
        max_length=100,
        description="パスワードは8文字以上で入力してください"
    )


# DBセッション依存性
def get_db():
    with Session(engine) as session:
        yield session


# アプリ起動時にテーブルを作成
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)