基礎から学ぶPython入門 90日コース | データベース・Web API － Day 71：FastAPI × SQLite

Python 90日で身につけるPython
スポンサーリンク
スポンサーリンク

Day 71：FastAPI × SQLite ― 「APIとデータベースを一本の線でつなぐ」

Day 71では、いよいよ FastAPI と SQLite を組み合わせて、 「ちゃんとデータベースに保存されるCRUD API」を作っていきます。

キーワードはこの3つです。

  • API
  • データベース接続
  • CRUD

ここまでで、

  • SQLiteの基礎（テーブル・SQL・CRUD）
  • ORMの基礎（モデル・CRUD）
  • FastAPIの基礎（GET / POST・リクエストボディ・バリデーション・CRUD API）

と、必要なピースはほぼ揃いました。

今日はそれらを一本の線でつなぎ、

「APIからリクエストが来る」 → 「SQLiteに保存・取得・更新・削除する」 → 「結果をJSONで返す」

という、Web/API開発の王道パターンを体験していきます。

全体像を決める ― 「何を管理するAPIにするか」

題材：シンプルな「顧客管理API」

Day 66で作った顧客管理CLIアプリを、 今日は Web API版 にしていきます。

扱う情報はシンプルです。

  • id：主キー（自動採番）
  • name：顧客名（必須）
  • email：メールアドレス（必須・重複なし）
  • phone：電話番号（任意）
  • note：メモ（任意）

これをSQLiteに保存しつつ、 FastAPIでCRUD APIを提供する、という構成にします。

データベース接続の準備 ― SQLiteとSQLAlchemyを使う

必要なライブラリのインストール

ORMとして SQLAlchemy を使います。

bash

pip install fastapi "uvicorn[standard]" sqlalchemy
  • fastapi：APIフレームワーク
  • uvicorn：サーバー
  • sqlalchemy：ORM（SQLiteとの橋渡し役）

データベース接続とベースクラス

# day71_fastapi_sqlite.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.orm import declarative_base, sessionmaker, Session

# FastAPIアプリケーション
app = FastAPI()

# SQLAlchemyのベースクラス（全モデルの親）
Base = declarative_base()

# SQLiteへの接続設定
DATABASE_URL = "sqlite:///day71_customers.db"

# エンジン（DBとの接続オブジェクト）を作成
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLite＋マルチスレッド対策
)

# セッション工場を作成
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Python

ここでのポイントは、

  • create_engine("sqlite:///...") でSQLiteファイルに接続する
  • SessionLocal で「DBとの会話窓口」を作る
  • check_same_thread=False は、FastAPIのようなマルチスレッド環境でSQLiteを使うための設定

というところです。

モデル定義 ― SQLiteのテーブルをPythonクラスで表す

顧客テーブルのモデル（SQLAlchemy）

class Customer(Base):
    """
    customers テーブルに対応するSQLAlchemyモデルです。
    テーブルのカラムと制約をここで定義します。
    """

    __tablename__ = "customers"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), nullable=False, unique=True, index=True)
    phone = Column(String(50), nullable=True)
    note = Column(Text, nullable=True)
Python
  • __tablename__：テーブル名
  • Column(...)：カラム定義
  • primary_key=True：主キー
  • nullable=False：NOT NULL
  • unique=True：UNIQUE制約
  • index=True：インデックス作成

テーブルを作成する

# モデル定義に基づいて、SQLite上にテーブルを作成します。
Base.metadata.create_all(bind=engine)
Python

これで、day71_customers.db に customers テーブルが作られます。

Pydanticモデル ― APIの入出力を表現する

リクエスト用・レスポンス用のモデル

class CustomerBase(BaseModel):
    """
    顧客情報の共通部分を表すPydanticモデルです。
    APIの入出力で使う「形」を定義します。
    """

    name: str
    email: EmailStr
    phone: str | None = None
    note: str | None = None


class CustomerCreate(CustomerBase):
    """
    顧客作成時のリクエストボディ用モデルです。
    id はサーバー側で採番するため含めません。
    """
    pass


class CustomerRead(CustomerBase):
    """
    顧客情報をレスポンスとして返すためのモデルです。
    DB側で管理している id を含みます。
    """

    id: int

    class Config:
        orm_mode = True
        # orm_mode=True にすることで、
        # SQLAlchemyモデルから直接このPydanticモデルに変換できるようになります。
Python

ここでのポイントは、

  • CustomerBase：共通項目
  • CustomerCreate：作成時の入力
  • CustomerRead：レスポンス用（id付き）
  • orm_mode = True：SQLAlchemyモデルをそのままPydanticに変換できるようにする設定

です。

セッション取得のヘルパー ― 「毎回ちゃんとDBとつなぐ」

def get_db() -> Session:
    """
    SQLAlchemyのセッション（DBとの会話窓口）を取得するためのヘルパー関数です。
    FastAPIのエンドポイント内で呼び出して使います。
    """

    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
Python

実際のFastAPIでは Depends(get_db) を使う形がよく登場しますが、 ここでは初心者向けに、明示的に db = SessionLocal() を呼ぶスタイル で進めます。

Create ― 顧客を作成するAPI（POST）

POST /customers

@app.post("/customers", response_model=CustomerRead)
def create_customer(customer_in: CustomerCreate):
    """
    顧客を新規作成するためのPOST APIです。

    - リクエストボディとして CustomerCreate を受け取ります。
    - SQLAlchemyの Customer モデルに変換してDBに保存します。
    - 保存された顧客情報を CustomerRead として返します。
    """

    db = SessionLocal()

    # 同じメールアドレスが既に登録されていないかチェックします。
    existing = db.query(Customer).filter(Customer.email == customer_in.email).first()
    if existing is not None:
        db.close()
        raise HTTPException(status_code=400, detail="Email already registered")

    # Customer モデルのインスタンスを作成します。
    customer = Customer(
        name=customer_in.name,
        email=customer_in.email,
        phone=customer_in.phone,
        note=customer_in.note,
    )

    # セッションに追加してコミットします。
    db.add(customer)
    db.commit()
    db.refresh(customer)  # DB側で確定した値（idなど）を反映します。

    db.close()

    # Pydanticモデル（CustomerRead）として返します。
    return customer
Python

ここでの流れは、

  1. CustomerCreate を受け取る
  2. メールアドレスの重複チェック
  3. Customer モデルを作成
  4. db.add() → db.commit() で保存
  5. db.refresh() で id を反映
  6. customer をそのまま返す（orm_mode=True により自動変換）

という、Create＋バリデーションの基本パターン です。

Read ― 顧客一覧・顧客詳細を取得するAPI（GET）

GET /customers ― 一覧取得

@app.get("/customers", response_model=list[CustomerRead])
def list_customers():
    """
    顧客一覧を取得するためのGET APIです。

    - DBから全顧客を取得します。
    - CustomerRead のリストとして返します。
    """

    db = SessionLocal()
    customers = db.query(Customer).order_by(Customer.id.asc()).all()
    db.close()

    return customers
Python

GET /customers/{customer_id} ― 1件取得

@app.get("/customers/{customer_id}", response_model=CustomerRead)
def get_customer(customer_id: int):
    """
    指定したIDの顧客を取得するためのGET APIです。

    - パスパラメータ customer_id を受け取ります。
    - DBから該当顧客を検索します。
    - 見つからなければ 404 を返します。
    """

    db = SessionLocal()
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    db.close()

    if customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")

    return customer
Python

Readでは、

  • query(Customer).all() で一覧
  • query(Customer).filter(...).first() で1件取得
  • 見つからない場合は HTTPException で404

という流れを押さえます。

Update ― 顧客情報を更新するAPI（PUT）

PUT /customers/{customer_id}

@app.put("/customers/{customer_id}", response_model=CustomerRead)
def update_customer(customer_id: int, customer_in: CustomerCreate):
    """
    指定したIDの顧客情報を更新するためのPUT APIです。

    - パスパラメータ customer_id で対象を指定します。
    - リクエストボディ customer_in で新しい内容を受け取ります。
    - 該当顧客が存在しなければ 404 を返します。
    - メールアドレスの重複があれば 400 を返します。
    """

    db = SessionLocal()
    customer = db.query(Customer).filter(Customer.id == customer_id).first()

    if customer is None:
        db.close()
        raise HTTPException(status_code=404, detail="Customer not found")

    # メールアドレスを変更する場合、他の顧客と重複していないかチェックします。
    if customer.email != customer_in.email:
        existing = db.query(Customer).filter(Customer.email == customer_in.email).first()
        if existing is not None:
            db.close()
            raise HTTPException(status_code=400, detail="Email already registered")

    # フィールドを更新します。
    customer.name = customer_in.name
    customer.email = customer_in.email
    customer.phone = customer_in.phone
    customer.note = customer_in.note

    db.commit()
    db.refresh(customer)
    db.close()

    return customer
Python

Updateでは、

  • 対象を取得して存在チェック
  • メールアドレスの重複チェック
  • フィールドを上書き
  • commit()＋refresh() で反映

という、慎重な更新の流れ を体験します。

Delete ― 顧客を削除するAPI（DELETE）

DELETE /customers/{customer_id}

@app.delete("/customers/{customer_id}")
def delete_customer(customer_id: int):
    """
    指定したIDの顧客を削除するためのDELETE APIです。

    - パスパラメータ customer_id で対象を指定します。
    - 該当顧客が存在しなければ 404 を返します。
    - 削除後、簡単なメッセージを返します。
    """

    db = SessionLocal()
    customer = db.query(Customer).filter(Customer.id == customer_id).first()

    if customer is None:
        db.close()
        raise HTTPException(status_code=404, detail="Customer not found")

    db.delete(customer)
    db.commit()
    db.close()

    return {"message": f"Customer {customer_id} deleted"}
Python

Deleteでは、

  • 対象を取得して存在チェック
  • db.delete() → db.commit() で削除
  • 成功メッセージを返す

という、シンプルだけれど重要な流れを押さえます。

Day 71ミニテンプレート ― FastAPI × SQLite CRUDひとまとめ

ここまでのコードを、ひとつのファイルにまとめたテンプレートです。

# day71_fastapi_sqlite_template.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.orm import declarative_base, sessionmaker, Session

app = FastAPI()

Base = declarative_base()
DATABASE_URL = "sqlite:///day71_customers.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)


class Customer(Base):
    __tablename__ = "customers"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), nullable=False, unique=True, index=True)
    phone = Column(String(50), nullable=True)
    note = Column(Text, nullable=True)


Base.metadata.create_all(bind=engine)


class CustomerBase(BaseModel):
    name: str
    email: EmailStr
    phone: str | None = None
    note: str | None = None


class CustomerCreate(CustomerBase):
    pass


class CustomerRead(CustomerBase):
    id: int

    class Config:
        orm_mode = True


@app.get("/")
def read_root():
    return {"message": "Day 71: FastAPI × SQLite CRUD API へようこそ"}


@app.post("/customers", response_model=CustomerRead)
def create_customer(customer_in: CustomerCreate):
    db: Session = SessionLocal()

    existing = db.query(Customer).filter(Customer.email == customer_in.email).first()
    if existing is not None:
        db.close()
        raise HTTPException(status_code=400, detail="Email already registered")

    customer = Customer(
        name=customer_in.name,
        email=customer_in.email,
        phone=customer_in.phone,
        note=customer_in.note,
    )

    db.add(customer)
    db.commit()
    db.refresh(customer)
    db.close()

    return customer


@app.get("/customers", response_model=list[CustomerRead])
def list_customers():
    db: Session = SessionLocal()
    customers = db.query(Customer).order_by(Customer.id.asc()).all()
    db.close()
    return customers


@app.get("/customers/{customer_id}", response_model=CustomerRead)
def get_customer(customer_id: int):
    db: Session = SessionLocal()
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    db.close()

    if customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")

    return customer


@app.put("/customers/{customer_id}", response_model=CustomerRead)
def update_customer(customer_id: int, customer_in: CustomerCreate):
    db: Session = SessionLocal()
    customer = db.query(Customer).filter(Customer.id == customer_id).first()

    if customer is None:
        db.close()
        raise HTTPException(status_code=404, detail="Customer not found")

    if customer.email != customer_in.email:
        existing = db.query(Customer).filter(Customer.email == customer_in.email).first()
        if existing is not None:
            db.close()
            raise HTTPException(status_code=400, detail="Email already registered")

    customer.name = customer_in.name
    customer.email = customer_in.email
    customer.phone = customer_in.phone
    customer.note = customer_in.note

    db.commit()
    db.refresh(customer)
    db.close()

    return customer


@app.delete("/customers/{customer_id}")
def delete_customer(customer_id: int):
    db: Session = SessionLocal()
    customer = db.query(Customer).filter(Customer.id == customer_id).first()

    if customer is None:
        db.close()
        raise HTTPException(status_code=404, detail="Customer not found")

    db.delete(customer)
    db.commit()
    db.close()

    return {"message": f"Customer {customer_id} deleted"}
Python

このファイルを保存して、

bash

uvicorn day71_fastapi_sqlite_template:app --reload

と実行すれば、

  • POST /customers
  • GET /customers
  • GET /customers/{id}
  • PUT /customers/{id}
  • DELETE /customers/{id}

という、FastAPI × SQLite のCRUD API を一通り試せます。

Day 71のまとめ ― 「APIとDBがつながった瞬間、サービスの“芯”が生まれる」

今日の主役は、

  • API：外からのリクエストを受け取り、JSONで応答する窓口
  • データベース接続：SQLiteとSQLAlchemyで、データを永続化する仕組み
  • CRUD：Create / Read / Update / Delete をAPI経由で行う流れ

でした。

FastAPIとSQLiteを組み合わせることで、

  • ブラウザや他のサービスから送られてきたデータを
  • バリデーションしつつ受け取り
  • データベースに保存し
  • 必要なときに取り出し、更新し、削除する

という、「サービスの芯」となる動きが、 ひとつのコードベースの中で完結するようになりました。

Day 71で体験したこの構成は、 少しスケールさせれば、そのまま「本番サービス」の土台にもなり得るものです。

ここから先は、

  • 認証・認可
  • ログ・エラー処理
  • テストコード
  • デプロイ（クラウドへの公開）

など、さらに広い世界が待っています。

でも、そのどれもが、 今日つないだ 「API × データベース × CRUD」 の上に乗っかっていくものです。

静かだけれど、確かな一本の線が、 Day 71であなたの中に引かれたはずです。

スポンサーリンク
タイトルとURLをコピーしました