ChatGPT先生と挑むFastAPI #6｜PostgreSQLとSQLAlchemy

それではSTEP6を始めましょう！

STEP6では以下の内容を学習します。

まず、この3つはそれぞれ役割がまったく違います。

クライアント
    ↓ HTTPリクエスト
FastAPI
    ↓
SQLAlchemy
    ↓
PostgreSQL

大まかにはこのような関係です。

FastAPIはこれまで使ってきたとおり、Web APIを作るためのフレームワークです。

• HTTPリクエストを受け取る
• パラメータを検証する
• 処理を実行する
• HTTPレスポンスを返す

という部分を担当します。

FastAPI自身がデータを永続保存してくれるわけではありません。

PostgreSQLは実際にデータを保存するデータベースです。

これまで学習で、以下のようなPython上のデータを使っていました。このデータはアプリケーションを停止すると消えてしまいます。

products = [
    {"id": 1, "name": "Pen", "price": 100},
]

STEP 6からは、以下のようにPostgreSQLへ保存します。

products テーブル

id | name | price
---+------+------
1  | Pen  | 100

これによってFastAPIを再起動してもデータが残ります。

ここが今回のSTEPで一番重要です。

SQLAlchemyは、Pythonからデータベースを操作するためのライブラリです。

SQLを直接書くなら、次のようになります。

SELECT *
FROM products
WHERE id = 1;

SQLAlchemyのORMを使うことで、Pythonのクラスやオブジェクトを利用してDBを操作できます。

イメージとしては、

product = Product(
    name="Pen",
    price=100,
)

というPythonのオブジェクトを、

productsテーブル

id | name | price
---+------+------
1  | Pen  | 100

というDBのレコードへ対応させます。

このPythonのオブジェクト ↔ DBのテーブル・レコードを対応付けて扱う仕組みが ORM（Object-Relational Mapping） です。

DRF経験と結び付けるなら、DjangoでProduct.objects.get(id=1)のようにModelを通してDBを操作していたのと、考え方としてはかなり近いです。

ただし大きな違いとして、DjangoではORMがフレームワークに組み込まれていますが、

Django
 └─ Django ORM

FastAPIにはORMが組み込まれていません。

そのため今回は、

FastAPI
    ↓
SQLAlchemy
    ↓
PostgreSQL

という組み合わせを自分たちで構成します。

STEP 5ではRouterやDependsを使って構成を整理しました。

今回DBを導入すると、たとえば最終的には概念的に、以下のような流れになります。

API
 ↓
Session
 ↓
SQLAlchemy
 ↓
PostgreSQL

そしてFastAPIのDependsも、DBを操作するSessionをAPIへ渡す仕組みとして使えるようになります。

ここで押さえておきたいのは、「FastAPIはAPIを担当し、PostgreSQLはデータ保存を担当し、その間をSQLAlchemyがつなぐ」という役割分担です。

また、SQLAlchemyそのものがデータベースなのではなく、PostgreSQLなどのDBをPythonから扱うためのライブラリという点も重要です。

STEP 1の環境構築時点で、今後使うものをまとめて用意しているため、すでに以下の構成になっています。

Docker Compose
├─ app  ：FastAPI開発環境
└─ db   ：PostgreSQL 17

FastAPI側からPostgreSQLへ接続するときは、以下を使用する構成です。今回は新しくPostgreSQLをインストールするのではなく、STEP 1で作ったDB環境が正常に動いていることを確認します。

ホスト名: db
ポート:   5432

VS CodeのDev Container内ではなく、Windows側のPowerShellまたはWSLで、FastAPIプロジェクトのルートディレクトリを開いてください。

そこで次を実行します。

docker compose ps

想定としては、appとdbの2つが表示されます。たとえばイメージはこんな感じです。

NAME        IMAGE          SERVICE   STATUS
xxxxxxxx    ...            app       Up ...
xxxxxxxx    postgres:17    db        Up ...

特に確認したいのは、SERVICEとdbの行があり、STATUSがUpになっていることです。

DBコンテナ内の環境変数をそのまま利用して、psqlからPostgreSQLへ接続してみます。

docker compose exec db sh -c 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB"'

成功すると、最後にxxxx=#のようなプロンプトが表示されます。

これは、PostgreSQLへ接続して、SQLを入力できる状態になったという意味です。

試しに、以下をを実行してください。PostgreSQLのバージョン情報が返ってくればOKです。

SELECT version();

結果表示がページャーで開いた場合はqキー押下でpsqlのプロンプトに戻れます。
psql自体を終了する場合は\q押下です。

今回ここで確認しているのは、以下の接続です。

PowerShell / WSL
       ↓
Docker
       ↓
dbコンテナ
       ↓
PostgreSQL

このあとSQLAlchemyを使う段階では、以下のような接続になります。

FastAPI（app）
       ↓
SQLAlchemy
       ↓
db:5432
       ↓
PostgreSQL

STEP 1ですでに導入してあるSQLAlchemyとpsycopgをここから実際に使い始めます。現在のプロジェクト構成に合わせて、DB接続処理は新しく app/database.py に分離します。

まず、現在.envには次の形式の接続先が設定されています。

DATABASE_URL=postgresql+psycopg://fastapi:fastapi@db:5432/fastapi_learning

この文字列を分解するとこうなります。

postgresql+psycopg:  DB + ドライバ
fastapi:             ユーザー
fastapi:             パスワード
db:                  host
5432:                port
fastapi_learning:    DB名

postgresql+psycopgは、以下を表しています。

• postgresql → 使用するDB
• psycopg → PostgreSQLへ接続するPythonドライバ

そして、ホスト名がdbなのはDocker Compose内でdbというサービス名を使っているためです。

app/database.pyを作成してください。

app/
├── main.py
├── dependencies.py
├── routers/
├── schemas/
└── database.py   ← 追加

内容はこちらです。

import os

from sqlalchemy import create_engine

# Docker Composeから渡されている環境変数から
# PostgreSQLへの接続URLを取得する
DATABASE_URL = os.environ["DATABASE_URL"]

# SQLAlchemyがDBへ接続するためのEngineを作成する
engine = create_engine(DATABASE_URL)

ここで重要なのが、engine = create_engine(DATABASE_URL)です。

SQLAlchemyでは、DBとの接続を管理する中心的なオブジェクトをEngineと呼びます。

イメージとしては以下です。

FastAPIの処理
     ↓
SQLAlchemy Engine
     ↓
psycopg
     ↓
PostgreSQL

ただし、create_engine(...)を実行した時点で毎回すぐPostgreSQLへ接続する、という意味ではありません。Engineは、「この設定でDBへ接続できるように準備しておくもの」と考えると分かりやすいです。

実際のDB接続は、後ほどEngineからConnectionやSessionが必要になったタイミングで行われます。

Djangoでいうと、この部分はフレームワーク側がかなり隠してくれています。SQLAlchemyでは、DB接続の土台となるEngineを自分で作るところが違いの一つですね。

VS CodeのDev Container内のターミナルで、プロジェクトルートから次を実行してください。

python - <<'PY'
from sqlalchemy import text

from app.database import engine

with engine.connect() as connection:
    result = connection.execute(text("SELECT 1"))
    print(result.scalar_one())
PY

正常なら、1と表示されます。

ここでは、engine.connect()によって実際にPostgreSQLへ接続し、SELECT 1を実行しています。

つまり1が返れば、以下の経路が正常につながったことを確認できます。

Python
 ↓
SQLAlchemy
 ↓
psycopg
 ↓
PostgreSQL

STEP 3ではPydanticのBaseModelを使って、APIで受け取るデータの形を定義しました。

今回作るSQLAlchemyのModelは、それとは役割が違います。

Pydantic Model：APIのリクエスト・レスポンスのデータ構造
SQLAlchemy Model：DBのテーブル構造

app/database.pyを次のように変更してください。

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase

# Docker Composeから渡されている環境変数から
# PostgreSQLへの接続URLを取得する
DATABASE_URL = os.environ["DATABASE_URL"]

# SQLAlchemyがDBへ接続するためのEngineを作成する
engine = create_engine(DATABASE_URL)


# すべてのSQLAlchemy Modelが継承する基底クラス
class Base(DeclarativeBase):
    pass

今回追加したのは、次の2つです。

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

このBaseを今後のSQLAlchemy Modelが継承します。

構成を次のようにします。

app/
├── main.py
├── database.py
├── dependencies.py
├── models/            ← 追加
│   ├── __init__.py   ← 追加
│   └── product.py    ← 追加
├── routers/
└── schemas/

app/models/product.pyを作成してください。

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Product(Base):
    # このModelが対応するDBテーブル名
    __tablename__ = "products"

    # 主キー
    id: Mapped[int] = mapped_column(primary_key=True)

    # 商品名
    name: Mapped[str] = mapped_column(String(100))

    # 価格
    price: Mapped[int] = mapped_column()

今回のポイントはこの3つです。

class Product(Base):

ProductはSQLAlchemyのModelです。先ほど作ったBaseを継承することで、「このクラスはSQLAlchemyが管理するDB Modelである」と認識されます。

次に、

__tablename__ = "products"

これは、このModelが対応するDBテーブル名です。つまり、「PythonのProductモデル」と「PostgreSQLのproducts」が対応することになります。

そして、

id: Mapped[int] = mapped_column(primary_key=True)

です。SQLAlchemy 2系では、このように、Mapped[型]とmapped_column(...)を使ってカラムを定義する書き方が基本です。

例えば、name: Mapped[str] = mapped_column(String(100))の場合は以下のようなカラムを表します。

カラム名：name
型：文字列
桁数：100文字

見た目が似ているので、ここは区別しておきましょう。

例えばPydanticなら、以下のように書いていました。

class ProductCreate(BaseModel):
    name: str
    price: int

これは、APIにどんなデータを渡せるかを表しています。

一方、今回の、

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[int] = mapped_column()

は、DBにどんな形でデータを保存するかを表します。したがって、今後は同じ「商品」であっても、以下の2種類が存在することになります。

schemas/
└── product.py
    → API側のデータ構造

models/
└── product.py
    → DB側のデータ構造

DRFでいうと、感覚的には以下に近いです。

Django Model ≒ SQLAlchemy Model
Serializer ≒ Pydantic Model

完全に同じ役割ではありませんが、整理するときにはかなり分かりやすい対応です。

ここは大事です。

今回、class Product(Base):を書いたからといって、PostgreSQLにproductsテーブルが自動作成されたわけではありません。

現時点では、

Python側
「productsテーブルはこういう構造にしたい」
        ↓
定義しただけ

です。実際にPostgreSQLへその変更を反映するのが、次の項目で扱うAlembicによるマイグレーションです。

Alembicは、SQLAlchemyのModel変更をDBのテーブル構造へ反映するためのマイグレーションツールです。

DRF/Django経験で対応させると、イメージはかなり近いです。

Django
Model変更
  ↓
python manage.py makemigrations
  ↓
python manage.py migrate
  ↓
DBへ反映

SQLAlchemy + Alembicでは、

SQLAlchemy Model変更
  ↓
alembic revision --autogenerate
  ↓
alembic upgrade head
  ↓
DBへ反映

となります。

ただし、Djangoと違ってAlembicはFastAPIやSQLAlchemyに内蔵されているわけではなく、別のライブラリです。

プロジェクトルートにまだ、

• alembic.ini
• migrations/

が存在しないことを確認してから、以下を実行します。

alembic init migrations

すると、以下のようなファイルが生成されます。

プロジェクトルート/
├── alembic.ini
├── migrations/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
└── app/

次に、migrations/env.pyを編集します。生成されたenv.pyの上の方に、まず以下を追加します。

import os

from app.database import Base
from app.models.product import Product  # noqa: F401

Productを直接このファイル内で使っていないのにimportしている理由があります。

Alembicが、productsというModelが存在していることを認識し、Base.metadataへ登録されたテーブル情報を読み取れるようにするためです。

そのため、このimportは副作用を目的としたimportです。

次に、生成時には、target_metadata = Noneとなっている箇所があります。これを、以下のように変更します。

target_metadata = Base.metadata

これでAlembicは、

Base.metadata
    ↓
SQLAlchemy Modelのテーブル情報
    ↓
Product → productsテーブル

を読み取れるようになります。

同じmigrations/env.pyで、config = context.configの後あたりに、次を追加します。

# Docker Composeから渡されているDB接続URLを
# Alembicでも使用する
config.set_main_option(
    "sqlalchemy.url",
    os.environ["DATABASE_URL"],
)

つまりAlembicも、SQLAlchemy本体と同じDATABASE_URLを使うようにします。これによって、

FastAPI / SQLAlchemy
        ↓
DATABASE_URL
        ↓
PostgreSQL

Alembic
        ↓
DATABASE_URL
        ↓
PostgreSQL

と、接続情報を共通化できます。

alembic.iniへユーザー名やパスワードを直接書かなくて済むのもメリットです。

Dev Container内で次を実行します。

alembic revision --autogenerate -m "create products table"

ここで、--autogenerateを付けるのがポイントです。

Alembicが、「現在のDB構造」と「SQLAlchemy Model」を比較して、「DBにはproductsテーブルがないので作る必要がある」と判断してマイグレーションファイルを生成します。

成功すると、以下のようなファイルができます。

migrations/versions/
└── xxxxxxxxxxxx_create_products_table.py

生成されたマイグレーションには、おおむね次のような処理が入っているはずです。

def upgrade() -> None:
    op.create_table(
        "products",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(length=100), nullable=False),
        sa.Column("price", sa.Integer(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )

これは、マイグレーションを適用するとproductsテーブルを作るという意味です。

また、以下も生成されるはずです。

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

こちらは逆に、マイグレーションを戻す場合はproductsテーブルを削除するという処理です。

--autogenerateで作ったからそのまま信用するのではなく、生成内容を確認してから適用するのが実務でも大事です。

内容に問題がなければ、以下を実行します。

alembic upgrade head

headは、現在存在する最新のマイグレーションまで適用するという意味です。

ここで初めて、

SQLAlchemy Model
      ↓
Alembic migration
      ↓
PostgreSQL

と反映され、実際にproductsテーブルが作られます。

前と同じ方法でpsqlへ入り、

docker compose exec db sh -c 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB"'

テーブル一覧を確認します。

\dt

正常なら、少なくとも以下の2つが確認できるはずです。

• alembic_version
• products

alembic_versionはAlembic自身が、現在どのマイグレーションまで適用済みかを管理するためのテーブルです。

さらに、以下のコマンドでproductsテーブルの構造も確認できます。

\d products

今回定義した、以下のカラムがあれば成功です。

• id
• name
• price

今回の重要な以下の流れです。

SQLAlchemy Modelを変更
        ↓
alembic revision --autogenerate
        ↓
生成されたmigrationを確認
        ↓
alembic upgrade head
        ↓
PostgreSQLへ反映

ここから、これまでPython上のリストを参照していたGET APIを、SQLAlchemy経由でPostgreSQLから取得する処理へ変えていきます。

まず必要になるのがSessionです。前に作ったengineはDB接続の土台でしたが、実際のCRUDでは基本的にSessionを通してDBを操作します。

FastAPI
  ↓
Session
  ↓
Engine
  ↓
PostgreSQL

Sessionそのものの考え方は項目9で改めて整理するので、今は「APIのDB操作で使うもの」と捉えてください。

app/database.pyを次のように変更します。

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker


# Docker Composeから渡されている環境変数から
# PostgreSQLへの接続URLを取得する
DATABASE_URL = os.environ["DATABASE_URL"]

# SQLAlchemyがDBへ接続するためのEngineを作成する
engine = create_engine(DATABASE_URL)

# DB操作に使用するSessionを生成するための設定
SessionLocal = sessionmaker(bind=engine)


# すべてのSQLAlchemy Modelが継承する基底クラス
class Base(DeclarativeBase):
    pass

追加したのは、以下の部分です。

# DB操作に使用するSessionを生成するための設定
SessionLocal = sessionmaker(bind=engine)

SessionLocal自体がSessionではなく、Sessionを作るためのものです。

db = SessionLocal()とすると、実際のSessionが生成されます。

STEP 5で作ったapp/dependencies.pyに、次の関数を追加します。既存処理は消さず、そのまま残してください。

from collections.abc import Generator

from sqlalchemy.orm import Session

from app.database import SessionLocal


def get_db() -> Generator[Session, None, None]:
    # 1リクエストで使用するSessionを作成する
    db = SessionLocal()

    try:
        # APIへSessionを渡す
        yield db
    finally:
        # APIの処理終了後にSessionを閉じる
        db.close()

API側で、db: Session = Depends(get_db)とすると、

リクエスト
   ↓
get_db()
   ↓
Session生成
   ↓
APIへ渡す
   ↓
API処理
   ↓
finally
   ↓
Sessionを閉じる

という流れになります。

yieldを使うことで、API処理の前後に処理を挟めるわけですね。

今回はまず、商品を1件取得するAPIをDB参照に変えます。

app/routers/products.py側で、必要なimportを追加します。

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

from app.dependencies import get_db
from app.models.product import Product

GET処理は次の形になります。

@router.get("/{product_id}")
def get_product(
    product_id: int,
    db: Annotated[Session, Depends(get_db)],
):
    # 主キーを指定してproductsテーブルから商品を取得する
    product = db.get(Product, product_id)

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

    return {
        "id": product.id,
        "name": product.name,
        "price": product.price,
    }

今回の中心はここです。

product = db.get(Product, product_id)

たとえば、product_idが1なら、概念的には以下のSQLに相当します。

SELECT ...
FROM products
WHERE id = 1;

Django ORMでいう、Product.objects.get(pk=1)に近いですね。

まだDBを使ったPOST APIは作っていないため、今回は確認用としてPostgreSQLへ直接1件だけ登録します。

psqlで、以下を実行してください。

INSERT INTO products (name, price)
VALUES ('Pen', 100);

続けて、SELECT * FROM products;で以下のように登録されていることを確認します。

id | name | price
---+------+------
1  | Pen  | 100

そのあとFastAPIを起動してSwagger UIから、GET /products/1を実行します。

正常なら、以下のレスポンスが返ります。

{
  "id": 1,
  "name": "Pen",
  "price": 100
}

存在しないIDの場合は以下のレスポンスが返ります。

{
  "detail": "Product not found"
}

今回確認したいのは、以下の流れです。

GETリクエスト
   ↓
Depends(get_db)
   ↓
Session
   ↓
db.get(Product, product_id)
   ↓
SQLAlchemy
   ↓
PostgreSQL
   ↓
取得した商品をレスポンス

POSTでは次の流れでDBへ登録します。

リクエスト
   ↓
Pydantic Modelで受け取る
   ↓
SQLAlchemy Modelを作る
   ↓
Sessionへ追加
   ↓
commit
   ↓
PostgreSQLへ保存

既存のPydantic Modelはそのまま使います。たとえばSTEP 3で作ったリクエスト用Modelが、以下だったとします。

class ProductCreate(BaseModel):
    name: str
    price: int

これは今回も、APIから受け取るデータの仕様としてそのまま使用します。

一方、前回作った以下は、DBへ保存するデータの構造です。

ここでPydantic ModelとSQLAlchemy Modelがつながります。

既存のPOST APIを、次の形に変更してください。

@router.post(
    "/products",
    status_code=status.HTTP_201_CREATED,
)
def create_product(
    product_data: ProductCreate,
    db: Annotated[Session, Depends(get_db)],
):
    # リクエストデータをもとにSQLAlchemy Modelを作成する
    product = Product(
        name=product_data.name,
        price=product_data.price,
    )

    # 新しい商品をSessionの管理対象に追加する
    db.add(product)

    # DBへ変更を確定する
    db.commit()

    # DBで確定した最新の内容を取得する
    db.refresh(product)

    return {
        "id": product.id,
        "name": product.name,
        "price": product.price,
    }

今回の重要部分を順番に見てみます。

product = Product(
    name=product_data.name,
    price=product_data.price,
)

ここではまだDBへINSERTしていません。SQLAlchemyのProductオブジェクトをPython上で作っただけです。

という変換をしています。

db.add(product)では、このproductをDBへ登録する対象としてSessionに追加するという操作をしています。まだ、この時点では変更を確定していません。

db.commit()でトランザクションを確定します。つまり、実際に、

INSERT INTO products ...

に相当する処理がPostgreSQLへ反映されます。

db.refresh(product)でDBに保存された最新の状態を、productへ読み直します。

今回特に重要なのがidです。POSTリクエストでは、

{
  "name": "Notebook",
  "price": 500
}

しかしDBへ保存すると、PostgreSQL側で、id = 2などの主キーが決まります。

そこで、db.refresh(product)を行うことで、product.idにもDBで確定した値が反映されます。

Swagger UIでPOST /productsを実行します。

以下をリクエストボディで送ります。

{
  "name": "Notebook",
  "price": 500
}

成功すれば次のようなのようなレスポンスになります。

{
  "id": 2,
  "name": "Notebook",
  "price": 500
}

さらに先ほど作った、GET /products/2を実行してください。

同じ商品が取得できれば、

POST /products
      ↓
PostgreSQLへ保存
      ↓
GET /products/2
      ↓
PostgreSQLから取得

までつながったことになります。

ここまでで、

GET    → PostgreSQLから取得
POST   → PostgreSQLへ登録

までできました。今回は残っている、

PUT    → PostgreSQLのデータを更新
DELETE → PostgreSQLのデータを削除

へ置き換えます。APIの仕様自体を変えるのではなく、これまでPython上のデータを操作していた部分をSQLAlchemyへ変更するイメージです。

PUT APIを次のように変更します。

@router.put("/products/{product_id}")
def update_product(
    product_id: int,
    product_data: ProductUpdate,
    db: Annotated[Session, Depends(get_db)],
):
    # 更新対象の商品をDBから取得する
    product = db.get(Product, product_id)

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

    # リクエストされた内容でModelを更新する
    product.name = product_data.name
    product.price = product_data.price

    # DBへ変更を確定する
    db.commit()

    # DBで確定した最新の内容を取得する
    db.refresh(product)

    return {
        "id": product.id,
        "name": product.name,
        "price": product.price,
    }

ここでPOSTとの違いに注目してください。POSTでは新しいModelなので、db.add(product)が必要でした。

一方UPDATEでは、product = db.get(Product, product_id)で取得した時点で、このproductはすでにSessionの管理対象です。

そのため、

product.name = ...
product.price = ...
db.commit()

だけで変更を検知してUPDATEしてくれます。

つまり、UPDATEでは改めてdb.add(product)する必要はありません。

DELETEもまず対象を取得します。

@router.delete("/products/{product_id}")
def delete_product(
    product_id: int,
    db: Annotated[Session, Depends(get_db)],
):
    # 削除対象の商品をDBから取得する
    product = db.get(Product, product_id)

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

    # 商品を削除対象にする
    db.delete(product)

    # DBへ変更を確定する
    db.commit()

    return {
        "message": "Product deleted",
    }

DELETEでは、db.delete(product)で、このModelを削除対象にするとSessionへ伝えます。

そのあと、db.commit()でPostgreSQLへDELETEが反映されます。

流れは以下です。

db.get()
   ↓
削除対象を取得
   ↓
db.delete()
   ↓
db.commit()
   ↓
PostgreSQLから削除

今回までに使ったSQLAlchemyの操作を整理すると、次のようになります。

# CREATE
db.add(product)
db.commit()

# READ
product = db.get(Product, product_id)

# UPDATE
product.name = "New name"
db.commit()

# DELETE
db.delete(product)
db.commit()

Django ORMでやっていたCRUDを、SQLAlchemyではSessionを中心に操作しているところが大きな違いですね。

まず登録済みの商品に対してPUTを実行します。

PUT /products/2を以下のリクエストボディで実行します。

{
  "name": "New Notebook",
  "price": 600
}

更新後にGET /products/2を実行して、以下になっていることを確認します。

{
  "id": 2,
  "name": "New Notebook",
  "price": 600
}

次に、DELETE /products/2を実行します。

そのあと再び、GET /products/2を実行して、404 Product not foundになれば削除できています。

存在しないIDに対するPUT・DELETEも、どちらも404になることを確認してください。

まずSessionは、単純に「DBへの接続」そのものではありません。

SQLAlchemyでは、DBから取得したModelや変更内容を管理し、DBとの一連のやり取りをまとめる役割を持っています。

イメージは以下です。

FastAPIの1リクエスト
        ↓
      Session
        ↓
 ┌────────┐
 │ Product取得    │
 │ Product追加    │
 │ Product更新    │
 │ Product削除    │
 └────────┘
        ↓
      Engine
        ↓
   PostgreSQL

今回、db: Annotated[Session, Depends(get_db)]としているdbが、このSessionです。

たとえば、product = db.get(Product, 1)とすると、取得したproductはSessionの管理対象になります。

そのため、

product.price = 600
db.commit()

だけで更新できます。明示的に、db.update(product)のような処理をしていないのはそのためです。

Sessionが、「このProductは取得したときから値が変わった」と把握しているので、commit()時に必要なUPDATEを実行してくれます。

POSTで、db.add(product)と書きました。この時点では、このオブジェクトをDBへ保存する対象としてSessionで管理するという状態です。

そして、db.commit()によって変更を確定します。

DELETEも同じです。

db.delete(product)
db.commit()

という2段階になっています。つまり、以下のような考え方です。

add / Model変更 / delete
        ↓
Session上で変更を管理
        ↓
commit
        ↓
DBへ確定

トランザクションは、複数のDB操作をひとまとまりの処理として扱う仕組みです。

db.commit()は今回行ったDB変更を確定する操作です。

反対に、db.rollback()は今のトランザクションでまだ確定していない変更を取り消す操作です。確定後は基本的に、そのトランザクションを単純なrollback()で元に戻すことはできません。

例えば、次のような使い方があります。

try:
    db.add(product)
    db.commit()
except Exception:
    db.rollback()
    raise

実務ではDB更新中にエラーが発生した場合、Sessionを正常な状態へ戻す意味でもrollback()が重要になります。

今回は直接使っていませんが、Sessionを理解する上で知っておくと整理しやすいものがあります。db.flush()です。

大まかには、以下のような関係です。

add / update / delete
       ↓
flush
       ↓
SQLをDBへ送る
ただし、まだ確定しない
       ↓
commit
       ↓
確定

実際にはcommit()をすると、その前に必要なflushもSQLAlchemyが行ってくれます。

flush()は、DBにはSQLを実行したいが、まだトランザクションは確定したくないという場合に使えます。

今の段階では、

flush = DBへ反映する処理を進める
commit = トランザクションを確定する

くらいで十分です。

db.refresh(product)はトランザクションを確定する処理ではありません。

なので、以下にはそれぞれ別の役割があります。

db.commit()
db.refresh(product)

今回のget_db()は、以下のようなコードでした。

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()

    try:
        yield db
    finally:
        db.close()

これによって基本的に、次のような使い方になります。

リクエストA
↓
Session A
↓
終了したらclose

リクエストB
↓
Session B
↓
終了したらclose

Sessionをアプリ全体で1個作って使い回すのではなく、リクエスト単位でSessionを分けるのが重要です。

これなら、あるユーザーのDB操作途中の状態を別のリクエストと不用意に共有しません。

Django ORMではこの辺りをフレームワーク側がかなり隠してくれますが、SQLAlchemyでは、Sessionをどこで作り、どこで使い、どこで閉じるかを自分たちで意識する必要があります。

今回のポイントをまとめると、次のような関係です。

Session         → DB操作とModelの状態を管理する
add()           → 新規登録対象としてSessionに追加
Modelの値を変更 → Sessionが変更を追跡
delete()        → 削除対象としてSessionに登録
commit()        → トランザクションを確定
rollback()      → 未確定の変更を取り消す
refresh()       → DBからそのModelの最新状態を読み直す
close()         → Sessionを終了する

今回は、STEP 6で学んだ内容を組み合わせて、既存の商品を複製して新しい商品としてDBへ登録するAPIを自力で作ってみましょう。

次のAPIを実装してください。

POST /products/{product_id}/duplicate

＜要件＞

今回学んだ内容だけで実装できます。

1. product_idを使って元の商品をDBから取得する
2. 商品が存在しなければ404を返す
3. 取得した商品のnameとpriceを使って、新しいProductを作る
4. 新しい商品をSessionへ追加する
5. DBへ保存を確定する
6. DBで確定した新しい商品の状態を取得する
7. 新しく作成された商品のid / name / priceを返す

たとえばDBに、

id | name     | price
---+----------+------
1  | Notebook | 500

が存在するとき、POST /products/1/duplicateを実行すると、元の商品と同じname・priceを持つ別の商品を新規登録します。

結果は以下のようになります。

{
  "id": 2,
  "name": "Notebook",
  "price": 500
}

自力課題は問題なく完了です。

ちなみにレスポンスが200なのも今回の要件上は問題ありません。新規リソースを作るAPIなので、実務では201 Createdにする設計もありますが、これは今回の自力課題の合否には関係ありません。

はい、これでOKです。
Swagger UI側も201レスポンスを正式に認識できています。

自力課題はこれで完全に完了です。