FastAPI×SQLite×SQLAlchemyで始める簡単CRUD実装ガイド
📋 記事のポイント(サマリー)
- 誰向け?
- FastAPIの基礎(
Hello, World!
API)は動かせるけれど、データベース連携は未経験〜初心者 - SQLiteを用いて手軽にCRUD機能を試してみたいエンジニア/学習者
- FastAPIの基礎(
- 学べること
- SQLite+SQLAlchemyの環境構築手順
- SQLAlchemyモデル(ORM)定義の方法
- Pydanticモデルとの連携ポイント
- FastAPIでのCRUDエンドポイント実装(Create/Read/Update/Delete)
- トラブルシューティングと次のステップ案内
- 得られる効果
- 軽量データベースSQLiteでローカル開発環境を整備
- SQLAlchemyのORM機能を理解し、FastAPIと組み合わせた実践的APIを構築
- データ永続化の基礎を習得し、応用可能な土台ができる
🎯 対象読者
- 学生エンジニアAさん(22歳)
大学の授業でFastAPIを触った経験はあるが、データベースはこれから学びたい方 - 社会人エンジニアBさん(28歳)
個人プロジェクトでSQLiteベースのAPIを手早く立ち上げて、CRUDを実装したい方 - 趣味プログラマーCさん(35歳)
簡単なタスク管理アプリを自作し、データの永続化を試してみたい方
♿ アクセシビリティレベル
- 構造化:見出し・リストで論理的にセクション分けし、スクリーンリーダーで把握しやすい
- 読みやすさ:漢字・ひらがな・カタカナのバランスを工夫し、難解な用語には注釈を添付
- コード例:コメント付きの幅固定コードブロックで提示し、補足説明を併記
- 要約:各章末に「要点まとめ」を設置し、復習しやすい設計
- パーソナルタッチ:やわらかい表現で、学習意欲を刺激する語り口
1. はじめに:なぜSQLite+SQLAlchemy?
Webアプリ開発でまず始めやすいのが、サーバ容量をほとんど消費せず手軽に立ち上げられる SQLite と、Pythonの型ヒントを活かしてDB操作をシンプルにしてくれる SQLAlchemy ORM の組み合わせです。
FastAPIとの相性も良く、少ない設定でCRUD機能(データの作成・取得・更新・削除)が実現できます。
本記事では、SQLite+SQLAlchemyの環境構築から、FastAPIへの組み込み、実際のCRUDエンドポイント実装までをステップ・バイ・ステップで解説します。
要点まとめ
- SQLite:ゼロコンフィグで使える軽量DB
- SQLAlchemy ORM:クラス定義ベースでDB操作が直感的
- FastAPI連携:自動コミット/セッション管理で開発効率UP
2. 環境構築
2.1 プロジェクトフォルダと仮想環境
mkdir fastapi-sqlite-crud
cd fastapi-sqlite-crud
python3 -m venv .venv && source .venv/bin/activate
要点まとめ:仮想環境内で依存を管理し、他プロジェクトと衝突を避ける
2.2 必要パッケージのインストール
(.venv) pip install fastapi uvicorn sqlalchemy databases aiosqlite
- fastapi:Webフレームワーク
- uvicorn:ASGIサーバー
- sqlalchemy:ORMライブラリ
- databases:非同期DB接続ラッパー
- aiosqlite:SQLite用非同期ドライバ
要点まとめ:async対応の
databases
+aiosqlite
で、FastAPIの非同期機能と親和性を保つ
3. SQLAlchemyモデルとデータベース接続
3.1 ディレクトリ構成例
fastapi-sqlite-crud/
├─ main.py
├─ models.py
├─ schemas.py
├─ database.py
└─ requirements.txt
3.2 database.py
:DB接続設定
# database.py
from databases import Database
from sqlalchemy import create_engine, MetaData
DATABASE_URL = "sqlite:///./test.db"
database = Database(DATABASE_URL) # 非同期操作用
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
metadata = MetaData() # テーブル定義のメタデータ
要点まとめ:
sqlite:///./test.db
でローカルファイルDBを指定check_same_thread=False
:複数スレッドからの接続を許可
3.3 models.py
:ORMモデル定義
# models.py
from sqlalchemy import Table, Column, Integer, String
from database import metadata, engine
# テーブル定義
items = Table(
"items",
metadata,
Column("id", Integer, primary_key=True),
Column("title", String, nullable=False),
Column("description", String, nullable=True),
)
# テーブル作成(初回起動時)
metadata.create_all(engine)
要点まとめ:
Table
+Column
でシンプルにスキーマ定義metadata.create_all()
でDBファイルにテーブルを生成
3.4 schemas.py
:Pydanticモデル定義
# schemas.py
from pydantic import BaseModel
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass # そのまま継承
class Item(ItemBase):
id: int
class Config:
orm_mode = True # ORMからの返却時に対応
要点まとめ:
- リクエスト用とレスポンス用に分けると柔軟性UP
orm_mode = True
でSQLAlchemyモデルを直接変換可能
4. CRUDエンドポイント実装
4.1 main.py
:全体構造
# main.py
from fastapi import FastAPI, HTTPException
from typing import list
from database import database
from models import items
from schemas import Item, ItemCreate
app = FastAPI(title="SQLite+SQLAlchemy CRUD API")
# 起動・終了イベントでDB接続管理
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
要点まとめ:
startup
/shutdown
イベントで接続開始/終了を自動化
4.2 Create:データ生成
@app.post("/items/", response_model=Item)
async def create_item(item: ItemCreate):
query = items.insert().values(title=item.title, description=item.description)
last_record_id = await database.execute(query)
return {**item.dict(), "id": last_record_id}
items.insert()
でINSERT文を生成database.execute()
は実行後、INSERTしたIDを返す
要点まとめ:
response_model
で戻り値をPydanticモデル化- dict展開で簡潔にIDを付与
4.3 Read:一覧取得
@app.get("/items/", response_model=list[Item])
async def read_items():
query = items.select()
return await database.fetch_all(query)
要点まとめ:
fetch_all()
で全行を取得し、リストで返却
4.4 Read:単一取得
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
query = items.select().where(items.c.id == item_id)
item = await database.fetch_one(query)
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
要点まとめ:
- 存在チェックと404エラー返却を実装
fetch_one()
で単一行を取得
4.5 Update:更新
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemCreate):
query = items.update().where(items.c.id == item_id).values(
title=item.title, description=item.description
)
await database.execute(query)
return {**item.dict(), "id": item_id}
要点まとめ:
update()
でUPDATE文を生成- 更新後はリクエストデータとIDを組み合わせて返却
4.6 Delete:削除
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
query = items.delete().where(items.c.id == item_id)
await database.execute(query)
return {"ok": True}
要点まとめ:
delete()
でDELETE文を生成- 成功時はシンプルなステータス返却
5. テスト&実行確認
5.1 サーバー起動
uvicorn main:app --reload
5.2 Swagger UIで操作
- http://127.0.0.1:8000/docs にアクセスし、各エンドポイントを試してみましょう。
要点まとめ:
- ブラウザ上でCRUDの動作を即時に確認可能
- 入力例やレスポンス例も自動生成
6. よくあるトラブルと対処法
現象 | 対処法 |
---|---|
OperationalError: unable to open database file |
実行ディレクトリに書き込み権限があるか確認。<br>パスを絶対パスに変更する。 |
staledatabaseconnection |
connect_args={"check_same_thread": False} 設定を見直し、database の切断/再接続を適切に管理。 |
型変換エラー(例:list[Item] が未対応) |
Python3.9未満なら from typing import List → response_model=List[Item] に変更。 |
7. まとめと次のステップ
本記事では、FastAPIとSQLite+SQLAlchemyを組み合わせ、基本的なCRUD機能を一から実装する流れをご紹介しました。
- 環境構築:仮想環境と必要パッケージのインストール
- モデル定義:SQLAlchemy+Pydanticでデータ構造を明確化
- エンドポイント実装:Create/Read/Update/Deleteを非同期で実行
- 動作確認:Uvicorn起動&Swagger UIで手軽にテスト
次のステップとしては、以下をぜひお試しくださいね♡
- 認証機能導入:OAuth2/JWTでセキュアなAPIに進化
- リレーション対応:ユーザー/投稿などの1対多・多対多を扱う
- テスト自動化:
pytest
でエンドポイントのユニットテスト実装 - 本番DB移行:PostgreSQLやMySQLへのスケールアップ
SQLiteでのCRUD体験を踏み台に、より大規模なデータベース連携へチャレンジしてみてくださいね✧