FastAPI×SQLite×SQLAlchemyで作るCRUD API入門

FastAPIでデータを保存するAPIを作るには、HTTPリクエストを受ける処理、入力と出力の検証、データベース操作を分けて考えると理解しやすくなります。

この記事では、SQLiteの1ファイルデータベースとSQLAlchemy ​ORMを使い、Itemを作成、取得、更新、削除する最小構成を組み立てます。

Python ​3.10以降を前提とし、同期版のSQLAlchemy ​Sessionに合わせてエンドポイントも通常のdefで実装します。

この記事で作るCRUD ​API

CRUDは、データを扱う四つの基本操作をまとめた呼び名です。

実装するHTTPエンドポイント

操作	HTTPメソッド	パス	役割
Create	POST	/items/	Itemを作成する
Read	GET	/items/	Itemの一覧を取得する
Read	GET	/items/{item_id}	IDを指定して1件取得する
Update	PUT	/items/{item_id}	既存のItemを更新する
Delete	DELETE	/items/{item_id}	既存のItemを削除する

SQLiteは、データを単一ファイルに保存できるデータベースです。

ORMは、データベースの行をPythonオブジェクトとして扱うための仕組みで、この例ではSQLAlchemy ​ORMを使います。

PydanticモデルはAPIが受け取る値と返す値の形を定義し、SQLAlchemyモデルはデータベースに保存する形を定義します。

環境を準備する

仮想環境を作成する

仮想環境を使うと、プロジェクトごとにPythonパッケージを分けて管理できます。

macOSまたはLinuxでは、次のコマンドを実行します。

mkdir ​fastapi-sqlite-crud
cd ​fastapi-sqlite-crud
python3 ​-m ​venv ​.venv
source ​.venv/bin/activate

Windows ​PowerShellでは、次のコマンドを実行します。

mkdir ​fastapi-sqlite-crud
cd ​fastapi-sqlite-crud
py ​-m ​venv ​.venv
.\.venv\Scripts\Activate.ps1

必要なパッケージをインストールする

python ​-m ​pip ​install ​fastapi ​uvicorn ​sqlalchemy

• FastAPI：HTTPリクエストを受け取り、APIの入出力を検証します。
• Uvicorn：FastAPIアプリケーションを動かすASGIサーバーです。
• SQLAlchemy：データベース接続、トランザクション、ORMを担当します。

ファイルを分ける

fastapi-sqlite-crud/
├── ​main.py
├── ​database.py
├── ​models.py
└── ​schemas.py

database.pyは接続設定、models.pyはテーブル定義、schemas.pyはAPIの入出力、main.pyはエンドポイントを担当します。

データベース接続を設定する

database.pyには、SQLAlchemyのEngineとSessionを作る処理を書きます。

# ​database.py
from ​collections.abc ​import ​Generator

from ​sqlalchemy ​import ​create_engine
from ​sqlalchemy.orm ​import ​Session, ​sessionmaker

DATABASE_URL ​= ​"sqlite:///./items.db"

engine ​= ​create_engine(
 ​ ​ ​ ​DATABASE_URL,
 ​ ​ ​ ​connect_args={"check_same_thread": ​False},
)
SessionLocal ​= ​sessionmaker(
 ​ ​ ​ ​bind=engine,
 ​ ​ ​ ​expire_on_commit=False,
)

def ​get_db() ​-> ​Generator[Session, ​None, ​None]:
 ​ ​ ​ ​with ​SessionLocal() ​as ​session:
 ​ ​ ​ ​ ​ ​ ​ ​yield ​session

Engineはデータベース接続を管理する入口で、Sessionは検索や保存をひとまとまりの処理として扱う窓口です。

get_db()はリクエストごとにSessionを渡し、処理後に閉じます。

check_same_thread=FalseはFastAPIからSQLiteを扱う構成に合わせた設定ですが、Sessionまで複数のリクエストで共有する指定ではありません。

SQLAlchemy ​ORMモデルを定義する

models.pyでは、itemsテーブルとPythonのItemクラスを対応させます。

# ​models.py
from ​sqlalchemy ​import ​String
from ​sqlalchemy.orm ​import ​DeclarativeBase, ​Mapped, ​mapped_column

class ​Base(DeclarativeBase):
 ​ ​ ​ ​pass

class ​Item(Base):
 ​ ​ ​ ​__tablename__ ​= ​"items"

 ​ ​ ​ ​id: ​Mapped[int] ​= ​mapped_column(primary_key=True)
 ​ ​ ​ ​title: ​Mapped[str] ​= ​mapped_column(String, ​nullable=False)
 ​ ​ ​ ​description: ​Mapped[str ​| ​None] ​= ​mapped_column(String, ​nullable=True)

Mappedの型注釈は、Python上の型とデータベース列の対応を示します。

idは主キー、titleは必須、descriptionはNULLを許す列です。

PydanticでAPIの入出力を定義する

schemas.pyでは、作成と更新で受け取るItemCreateと、返却に使うItemReadを分けます。

# ​schemas.py
from ​pydantic ​import ​BaseModel, ​ConfigDict

class ​ItemBase(BaseModel):
 ​ ​ ​ ​title: ​str
 ​ ​ ​ ​description: ​str ​| ​None ​= ​None

class ​ItemCreate(ItemBase):
 ​ ​ ​ ​pass

class ​ItemRead(ItemBase):
 ​ ​ ​ ​model_config ​= ​ConfigDict(from_attributes=True)

 ​ ​ ​ ​id: ​int

from_attributes=Trueにより、PydanticはSQLAlchemyのItemオブジェクトからレスポンスを組み立てられます。

リクエストとレスポンスの型設計を詳しく確認したい場合は、FastAPIとPydantic ​v2でリクエストとレスポンスを定義する方法も参照してください。

CRUDエンドポイントを実装する

main.pyにテーブルの初期化と五つのエンドポイントをまとめます。

# ​main.py
from ​contextlib ​import ​asynccontextmanager
from ​typing ​import ​Annotated

from ​fastapi ​import ​Depends, ​FastAPI, ​HTTPException, ​Query
from ​sqlalchemy ​import ​select
from ​sqlalchemy.orm ​import ​Session

from ​database ​import ​engine, ​get_db
from ​models ​import ​Base, ​Item
from ​schemas ​import ​ItemCreate, ​ItemRead

@asynccontextmanager
async ​def ​lifespan(app: ​FastAPI):
 ​ ​ ​ ​Base.metadata.create_all(bind=engine)
 ​ ​ ​ ​yield
 ​ ​ ​ ​engine.dispose()

app ​= ​FastAPI(
 ​ ​ ​ ​title="SQLite ​and ​SQLAlchemy ​CRUD ​API",
 ​ ​ ​ ​lifespan=lifespan,
)

SessionDep ​= ​Annotated[Session, ​Depends(get_db)]

@app.post("/items/", ​response_model=ItemRead, ​status_code=201)
def ​create_item(payload: ​ItemCreate, ​db: ​SessionDep):
 ​ ​ ​ ​item ​= ​Item(**payload.model_dump())
 ​ ​ ​ ​db.add(item)
 ​ ​ ​ ​db.commit()
 ​ ​ ​ ​db.refresh(item)
 ​ ​ ​ ​return ​item

@app.get("/items/", ​response_model=list[ItemRead])
def ​read_items(
 ​ ​ ​ ​db: ​SessionDep,
 ​ ​ ​ ​offset: ​Annotated[int, ​Query(ge=0)] ​= ​0,
 ​ ​ ​ ​limit: ​Annotated[int, ​Query(ge=1, ​le=100)] ​= ​20,
):
 ​ ​ ​ ​statement ​= ​select(Item).offset(offset).limit(limit)
 ​ ​ ​ ​return ​list(db.scalars(statement).all())

@app.get("/items/{item_id}", ​response_model=ItemRead)
def ​read_item(item_id: ​int, ​db: ​SessionDep):
 ​ ​ ​ ​item ​= ​db.get(Item, ​item_id)
 ​ ​ ​ ​if ​item ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(status_code=404, ​detail="Item ​not ​found")
 ​ ​ ​ ​return ​item

@app.put("/items/{item_id}", ​response_model=ItemRead)
def ​update_item(item_id: ​int, ​payload: ​ItemCreate, ​db: ​SessionDep):
 ​ ​ ​ ​item ​= ​db.get(Item, ​item_id)
 ​ ​ ​ ​if ​item ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(status_code=404, ​detail="Item ​not ​found")

 ​ ​ ​ ​item.title ​= ​payload.title
 ​ ​ ​ ​item.description ​= ​payload.description
 ​ ​ ​ ​db.commit()
 ​ ​ ​ ​db.refresh(item)
 ​ ​ ​ ​return ​item

@app.delete("/items/{item_id}")
def ​delete_item(item_id: ​int, ​db: ​SessionDep):
 ​ ​ ​ ​item ​= ​db.get(Item, ​item_id)
 ​ ​ ​ ​if ​item ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(status_code=404, ​detail="Item ​not ​found")

 ​ ​ ​ ​db.delete(item)
 ​ ​ ​ ​db.commit()
 ​ ​ ​ ​return ​{"ok": ​True}

lifespan内のcreate_all()は、アプリケーションの起動時に未作成のテーブルを用意します。

起動処理と終了処理の役割は、FastAPIのlifespanで共有リソースを管理する方法で詳しく解説しています。

Createの処理

model_dump()で入力値を辞書にし、SQLAlchemyのItemへ渡します。

commit()で変更を確定し、refresh()で採番されたIDを含む最新状態を読み直します。

Readの処理

一覧取得では、offsetとlimitを使って取得範囲を制限します。

1件取得ではSession.get()で主キーを検索し、見つからない場合は404を返します。

UpdateとDeleteの処理

更新と削除でも対象を先に検索し、存在する場合だけデータベースを変更します。

この確認がない実装では、存在しないIDに対して成功したような応答を返してしまうため、利用側が処理結果を判断できません。

サーバーを起動して動作を確認する

uvicorn ​main:app ​--reload

起動後にhttp://127.0.0.1:8000/docsを開くと、Swagger ​UIから各エンドポイントを実行できます。

最初にPOST ​/items/でItemを作り、返されたIDを使って取得、更新、削除の順に確認すると、データの変化を追いやすくなります。

{
 ​ ​"title": ​"買い物リストを作る",
 ​ ​"description": ​"CRUD ​APIの動作確認"
}

エラーが出たときの確認箇所

よくある現象と確認箇所

現象	確認箇所
ModuleNotFoundError	仮想環境が有効か、同じ環境へ必要なパッケージをインストールしたかを確認します。
unable ​to ​open ​database ​file	起動したディレクトリと、そのディレクトリへの書き込み権限を確認します。
SQLiteのスレッドに関するエラー	create_engine()のconnect_argsと、Sessionをリクエストごとに生成しているかを確認します。
HTTP ​422	リクエスト本文がJSONか、titleが文字列で含まれているかをSwagger ​UIで確認します。
HTTP ​404	取得、更新、削除に指定したIDがデータベースに存在するかを確認します。

学習用構成から先へ進むときの確認事項

この構成は、CRUDとORMの流れを手元で確かめるための出発点です。

業務利用へ進める場合は、次の項目を要件に合わせて設計します。

• 空文字や文字数など、業務ルールに沿った入力検証
• 認証と権限管理
• 異常時のロールバックとエラー応答
• 自動テストとログ
• データ量と同時更新を踏まえたデータベース選定
• テーブル定義の変更履歴を管理するマイグレーション

create_all()だけでテーブル変更の履歴は管理できません。

スキーマを継続的に変更する段階では、FastAPI、SQLAlchemy、AlembicによるDBマイグレーションへ進むと、変更を安全に追跡する考え方を学べます。

FastAPIとSQLiteでCRUDを学ぶ要点

• PydanticモデルとSQLAlchemyモデルは、APIの入出力と保存形式という別の役割を持ちます。
• Sessionはリクエストごとに作成し、書き込み後はcommit()で確定します。
• 取得、更新、削除では、対象が存在しない場合に404を返します。
• Swagger ​UIを使うと、作成から削除までの流れをブラウザで確認できます。

この四点をコードと実行結果で結び付ければ、認証、関連テーブル、テスト、マイグレーションを学ぶための土台ができます。

この記事に関連する株式会社greedenの取り組み

SQLiteで試作したCRUDを業務で使える仕組みへ育てるには、要件整理、設計、テスト、運用までをつなぐ視点が必要です。株式会社greedenは、Webシステム開発を企画から保守改善まで一貫して支援します。