SQL (Relational) Databases¶

!!! info"情報" これらの情報は間もなく更新されます。 🎉

現在のバージョンは Pydantic v1、SQLAlchemy のバージョンは 2.0 未満を想定しています。

新しいドキュメントには Pydantic v2 が含まれ、<a href="https://sqlmodel.tiangolo.com/" class="external-link" target="_blank">SQLModel</a> (これも SQLAlchemy をベースにしています) が Pydantic v2 を使用するように更新され次第、SQLModel も使用される予定です。

FastAPI はあなたに SQL(relational)を使用することを要求しません。

But you can use any relational database that you want. しかし、あなたが望むどの SQL(relational)も利用することができます。

ここに SQLAlchemyを利用した例があります。

あなたは簡単に以下にあるような SQLAlchemy によってサポートされたどのデータベースも利用することが可能可能です。

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server, etc.

In this example, we'll use SQLite, because it uses a single file and Python has integrated support. So, you can copy this example and run it as is. この例では、SQLite を使用します。SQLiteは単一のファイルを使用し、Python に統合サポートが組み込まれているためです。そのため、この例をコピーしてそのまま実行できます。

後ほど、本番環境のアプリケーションでは、PostgreSQLのようなデータベースサーバーを使用したくなるかもしれません。

豆知識

FastAPIとPostgreSQLを使用した公式プロジェクトジェネレーターがあります。すべて Docker ベースで、フロントエンドやその他のツールも含まれています。 https://github.com/tiangolo/full-stack-fastapi-postgresql

備考

ほとんどのコードは、フレームワークに依存しない標準的な SQLAlchemy のコードであることに注意してください。

**FastAPI**特有のコードは常に最小限です

ORMs(オブジェクト関係マッピング)¶

FastAPIは、あらゆるデータベース、そしてデータベースと通信するためのあらゆるスタイルのライブラリと連携します。

一般的なパターンは、「ORM」: つまり「オブジェクト関係マッピング」ライブラリを使用することです。

ORM には、コード内の オブジェクト とデータベースのテーブル（リレーション）間で変換（「マッピング」）を行うためのツールがあります。

ORM を使用すると、通常は SQL データベース内のテーブルを表すクラスを作成します。クラスの各属性は列を表し、名前と型を持ちます。

例えば、Pet というクラスは、pets という SQL テーブルを表すことができます。

そして、そのクラスの インスタンス オブジェクトはそれぞれ、データベース内の行を表します。

例えば、orion_cat というオブジェクト (Pet のインスタンス) は、type という列に対応する orion_cat.type という属性を持つことができます。そして、その属性の値は、例えば "cat" となります。

これらの ORM には、テーブルやエンティティ間の接続や関係を作成するためのツールも用意されています。

このように、orion_cat.owner という属性を持つこともできます。そして、owner には、owners テーブルから取得された、このペットの飼い主のデータが含まれます。

つまり、orion_cat.owner.name は、このペットの飼い主の名前 (owners テーブルの name 列から取得) になります。

例えば "Arquilian" という値を持つことができます。

そして、ORM は、ペットオブジェクトからアクセスしようとしたときに、対応するテーブル owners から情報を取得するためのすべての処理を行います。

一般的な ORM には、例えば Django-ORM (Django フレームワークの一部)、SQLAlchemy ORM (SQLAlchemy の一部、フレームワークに依存しない)、Peewee (フレームワークに依存しない) などがあります。

ここでは SQLAlchemy ORM の使用方法を見ていきます。

同様の方法で、他の ORM を使用することもできます

豆知識

Peewee を使用した同様の記事が、このドキュメントにあります。

ファイル構造¶

これらの例では、my_super_project という名前のディレクトリがあり、その中に sql_app という名前のサブディレクトリがあるとします。その構造は次のとおりです。

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

__init__.py ファイルは空のファイルですが、これにより Python は、（Python ファイルである）すべてのモジュールを含むsql_appがパッケージであることを認識します。

では、各ファイル/モジュールが何をするのかを見ていきましょう。

`SQLAlchemy`をインストールする¶

最初にSQLAlchemyをインストールする必要があります。

$ pip install sqlalchemy

---> 100%

SQLAlchemy のパーツを作成する¶

このファイルを参照してください。 sql_app/database.py.

SQLAlchemy のパーツをインポートする¶

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

SQLAlchemy 用のデータベース URL を作成する¶

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

この例では、SQLite データベースに「接続」しています（SQLite データベースを含むファイルを開いています）。

ファイルは、sql_app.db というファイル内の、同じディレクトリにあります。

そのため、最後の部分は ./sql_app.db となっています。

PostgreSQL データベースを使用している場合は、次の行のコメントを外すだけです。

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

...そして、データベースのデータと認証情報で適宜変更してください（MySQL、MariaDB、その他のデータベースでも同様です）

豆知識

これは、別のデータベースを使用したい場合に変更する必要がある主要な行です。

SQLAlchemy `engine` を作成する¶

最初のステップは、SQLAlchemy の「エンジン」を作成することです。

この engine は、後で他の場所で使用します。

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

備考¶

議論:

connect_args={"check_same_thread": False}

...これは SQLiteにのみ必要です。ほかのデータベースでは必要ありません。

技術的な詳細

デフォルトでは、SQLite は 1 つのスレッドのみが自身と通信できるようにします。これは、各スレッドが独立したリクエストを処理すると想定しているためです。

これは、異なるもの（異なるリクエスト）に対して同じ接続を誤って共有することを防ぐためです。

しかし、FastAPI では、通常の関数 (def) を使用すると、複数のスレッドが同じリクエストに対してデータベースと対話する可能性があるため、 connect_args={"check_same_thread": False} を使用して SQLite にそれを許可するように指示する必要があります。

また、各リクエストが依存関係の中で独自のデータベース接続セッションを取得するようにするため、このデフォルトのメカニズムは必要ありません。

`SessionLocal` クラスを作成する¶

SessionLocal クラスの各インスタンスは、データベースセッションになります。クラス自体は、まだデータベースセッションではありません。

しかし、SessionLocal クラスのインスタンスを作成すると、このインスタンスが実際のデータベースセッションになります。

SQLAlchemy からインポートした Session と区別するために、SessionLocal という名前を付けています。

Session (SQLAlchemy からインポートしたもの) は後で使用します。

SessionLocal クラスを作成するには、sessionmaker 関数を使用します。

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

`Base` クラスを作成する¶

次に、クラスを返す declarative_base() 関数を使用します。

後で、このクラスを継承して、データベースモデルまたはクラス（ORM モデル）を作成します。

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Create the database models¶

このファイルを見てみましょう！ sql_app/models.py

`Base` クラスから SQLAlchemy モデルを作成する¶

先ほど作成した Base クラスを使用して、SQLAlchemy モデルを作成します。

豆知識

SQLAlchemy は、「モデル」という用語を使用して、データベースと対話するこれらのクラスやインスタンスを参照します。

しかし、Pydantic も「**モデル**」という用語を使用して、データの検証、変換、およびドキュメントのクラスとインスタンスという、別のものを指します。

database (上記の database.py ファイル) から Base をインポートします。

それを継承するクラスを作成します。

これらのクラスが SQLAlchemy モデルです。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

__tablename__ 属性は、これらの各モデルに対してデータベース内で使用するテーブル名を SQLAlchemy に伝えます。

モデルの属性/列を作成する¶

次に、すべてのモデル (クラス) 属性を作成します。

これらの属性はそれぞれ、対応するデータベーステーブル内の列を表します。

デフォルト値として、SQLAlchemy の Column を使用します。

そして、データベース内の型を定義する SQLAlchemy クラスの「型」(Integer、String、Boolean など) を引数として渡します。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

関係(relationship)の作成¶

次に、リレーションシップを作成します。

これには、SQLAlchemy ORM によって提供される relationship を使用します。

これは、多かれ少なかれ、「マジック」属性となり、このテーブルに関連する他のテーブルの値を含みます。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

my_user.items のように User 内の items 属性にアクセスすると、users テーブル内のこのレコードを指す外部キーを持つ Item SQLAlchemy モデル (items テーブルからの) のリストが格納されます。

my_user.items にアクセスすると、SQLAlchemy は実際に items テーブル内のデータベースからアイテムを取得し、ここに格納します。

また、Item 内の owner 属性にアクセスすると、users テーブルからの User SQLAlchemy モデルが含まれます。users テーブルからどのレコードを取得するかを知るために、外部キーを持つ owner_id 属性/列を使用します。

Create the Pydantic models¶

次にファイルを確認しましょう sql_app/schemas.py.

豆知識

SQLAlchemy の モデル と Pydantic の モデル の混 confusion を避けるために、SQLAlchemy モデルを格納するファイル models.py と、Pydantic モデルを格納するファイル schemas.py を作成します。

これらの Pydantic モデルは、多かれ少なかれ「スキーマ」（有効なデータの形状）を定義します。

これにより、両方を使い分けるときに混乱を避けることができます。

初期の Pydantic モデル / スキーマを作成する¶

データの作成や読み取り時に共通の属性を持つように、ItemBase と UserBase の Pydantic モデル (あるいは「スキーマ」と呼ぶことにしましょう) を作成します。

そして、それらを継承する ItemCreate と UserCreate を作成します（これにより、同じ属性を持ちます）。さらに、作成に必要な追加のデータ（属性）を追加します。

したがって、ユーザーを作成するときは、password も持ちます。

ただし、セキュリティ上の理由から、password は他の Pydantic モデル には含まれません。例えば、API からユーザーを読み取るときに、password は送信されません。

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


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


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

SQLAlchemy スタイルと Pydantic スタイル¶

SQLAlchemy モデル は、= を使用して属性を定義し、Column にパラメータとして型を渡すことに注意してください

name = Column(String)

一方、Pydantic モデル は、新しい型注釈構文/型ヒントを使用して、: で型を宣言します。

name: str

これらを覚えておいてください。そうすれば、= と : を使用するときに混乱することがなくなります。

読み取り/返却用の Pydantic モデル / スキーマを作成する¶

次に、データを読み取るとき、つまり API からデータを返すときに使用される Pydantic モデル (スキーマ) を作成します。

例えば、アイテムを作成する前は、それに割り当てられる ID はわかりませんが、アイテムを読み取るとき (API から返すとき) は、すでにその ID がわかっています。

同様に、ユーザーを読み取るときに、items にはこのユーザーに属するアイテムが含まれることを宣言できます。

これらのアイテムの ID だけでなく、アイテムを読み取るための Pydantic モデル (Item) で定義したすべてのデータも含まれます。

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


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


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

豆知識

ユーザーを読み取るとき (API から返すとき) に使用される Pydantic モデル である User には、password が含まれていないことに注意してください。

Pydantic の `orm_mode` を使用する¶

次に、読み取り用の Pydantic モデル である Item と User に、内部の Config クラスを追加します。

この Config クラスは、Pydantic に設定を提供するために使用されます。

Config クラスで、属性 orm_mode = True を設定します。

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


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


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

豆知識

orm_mode = True のように、= を使用して値を代入していることに注意してください。

これは、前の型宣言のように `:` を使用していません。

これは、型を宣言するのではなく、設定値を設定しています。

Pydantic の `orm_mode` は、`dict` ではなくても、ORM モデル (または属性を持つ任意のオブジェクト) であっても、Pydantic *モデル* にデータを読み取るように指示します。

このようにして、次のように、`dict` から `id` 値を取得しようとするだけでなく、

id = data["id"]

次のように、属性からも取得しようとします。

id = data.id

このようにして、Pydantic モデル は ORM と互換性を持つようになり、パス操作 の response_model 引数で宣言するだけで済みます。

データベースモデルを返すことができ、そこからデータを読み取ることができます。

ORM モードに関する技術的な詳細¶

SQLAlchemy をはじめとする多くの ORM は、デフォルトで「遅延読み込み」になっています。

これは、例えば、関係を持つデータを含む属性にアクセスしようとしない限り、データベースからそのデータを取得しないことを意味します。

例えば、items 属性にアクセスするとします。

current_user.items

すると、SQLAlchemy は items テーブルにアクセスして、このユーザーのアイテムを取得します。しかし、それまでは取得しません。

orm_mode を使用しないと、パス操作 から SQLAlchemy モデルを返しても、関係を持つデータは含まれません。

Pydantic モデルでそれらの関係を宣言していても、含まれません。

しかし、ORM モードを使用すると、Pydantic 自体が (dict を想定するのではなく) 属性から必要なデータにアクセスしようとするため、返したい特定のデータを宣言することができ、ORM からであっても、そのデータを取得することができます。

CRUD ユーティリティ¶

では、sql_app/crud.py ファイルを見てみましょう。

このファイルには、データベース内のデータを操作するための再利用可能な関数を記述します。

CRUD は、Create (作成)、Read (読み取り)、Update (更新)、Delete (削除) の頭文字をとったものです。

...ただし、この例では、作成と読み取りのみを行っています。

データの読み込み¶

sqlalchemy.orm から Session をインポートします。これにより、db パラメータの型を宣言し、関数内でより適切な型チェックと補完を行うことができるようになります。

models (SQLAlchemy モデル) と schemas (Pydantic モデル / スキーマ) をインポートします。

以下の機能を提供するユーティリティ関数を作成します。

ID とメールアドレスで 1 人のユーザーを読み取る。
複数のユーザーを読み取る。
複数のアイテムを読み取る。

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

豆知識

パス操作関数 から独立して、データベースとの対話 (ユーザーやアイテムの取得) のみを担当する関数を作成することで、複数の場所でより簡単に再利用できるようになり、また、単体テストを追加しやすくなります。

データの作成¶

ユーティリティ関数を書いてデータを作成する手順は以下の通りです。

データを使用して SQLAlchemy モデルのインスタンスを作成します。
インスタンスオブジェクトをデータベースセッションに追加します。
データベースへの変更をコミットします（これにより、変更が保存されます）。
インスタンスを更新します（これにより、インスタンスにデータベースからの新しいデータ（生成された ID など）が含まれます）。

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

情報

Pydantic v1 では、このメソッドは.dict()と呼ばれていました。Pydantic v2 では非推奨になりました（ただし、引き続きサポートされています）が、.model_dump()に名前が変更されました。

これらの例では、Pydantic v1 との互換性のために.dict()を使用していますが、Pydantic v2 を使用できる場合は、代わりに.model_dump()を使用する必要があります。

豆知識

Userの SQLAlchemy モデルには、パスワードの安全なハッシュバージョンを含むhashed_passwordが含まれています。

しかし、API クライアントが提供するのは元のパスワードなので、アプリケーションでそれを抽出してハッシュパスワードを生成する必要があります。

次に、値を含むhashed_password引数を渡して保存します。

注意

この例は安全ではありません。パスワードはハッシュ化されていません。

実際のアプリケーションでは、パスワードをハッシュ化し、プレーンテキストで保存しないようにする必要があります。

詳細については、チュートリアルのセキュリティセクションに戻ってください。

ここでは、データベースのツールとメカニズムのみに焦点を当てています。

豆知識

Item にキーワード引数を 1 つずつ渡して、各引数を Pydantic モデルから読み取るのではなく、Pydantic モデルのデータを dict に生成して、以下のようにしています。

`item.dict()`

そして以下のように、dictのキーと値のペアを、SQLAlchemyのItemにキーワード引数として渡しています。

`Item(**item.dict())`

そして、Pydantic *モデル*によって提供されない追加のキーワード引数`owner_id`を渡します。以下のようにです:

`Item(**item.dict(), owner_id=user_id)`

メインの FastAPI アプリ¶

そして、sql_app/main.pyファイルで、これまでに作成した他のすべての部分を統合して使用してみましょう。

データベーステーブルの作成¶

非常に単純な方法で、データベーステーブルを作成します。

Python 3.9+Python 3.8+

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from typing import List

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

Alembic メモ¶

通常、データベースを初期化する場合（テーブルの作成など）、Alembicを使用します。

また、Alembic は「マイグレーション」にも使用します（これは Alembic の主な仕事です）。

「マイグレーション」とは、SQLAlchemy モデルの構造を変更したり、新しい属性を追加したりなどした場合に、データベースにそれらの変更を複製したり、新しい列や新しいテーブルを追加したりするために必要な手順のセットです。

FastAPI プロジェクトで Alembic の例を見つけるには、プロジェクト生成 - テンプレートのテンプレートを使用します。具体的には、ソースコードのalembicディレクトリをご覧ください。

依存関係を作成します¶

これで、sql_app/database.pyファイルで作成したSessionLocalクラスを使用して、依存関係を作成します。

各リクエストに対して独立したデータベースセッション/接続(SessionLocal)が必要であり、そのセッションをリクエスト全体で使用し、リクエストが完了したらクローズする必要があります。

そして、次のリクエストのために新しいセッションが作成されます。

そのため、依存関係yieldに関するセクションで説明したように、yieldを使用した新しい依存関係を作成します。

私たちの依存関係は、単一のリクエストで使用され、リクエストが終了したらクローズされる新しい SQLAlchemy のSessionLocalを作成します。

Python 3.9+Python 3.8+

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from typing import List

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

情報

SessionLocal()の作成とリクエストの処理をtryブロックに入れます。

そして、finallyブロックでクローズします。

これにより、リクエスト後にデータベースセッションが常にクローズされるようにします。リクエストの処理中に例外が発生した場合でもです。

しかし、終了コード（`yield`の後）から別の例外を発生させることはできません。 [依存関係`yield`と`HTTPException`](dependencies/dependencies-with-yield.md#dependencies-with-yield-and-httpexception){.internal-link target=_blank}で詳しく説明されています。

そして、パスオペレーション関数で依存関係を使用する場合は、SQLAlchemy から直接インポートしたSession型で宣言します。

これにより、パスオペレーション関数内でより良いエディターサポートが得られます。なぜなら、エディターはdbパラメーターがSession型であることを認識しているからです。

Python 3.9+Python 3.8+

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from typing import List

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

技術的な詳細

パラメーターdbは実際にはSessionLocal型ですが、このクラス（sessionmaker()で作成された）は SQLAlchemy のSessionの「プロキシ」なので、エディターは実際にはどのメソッドが提供されているかを知りません。

しかし、Sessionとして型を宣言することにより、エディターは使用可能なメソッド（.add()、.query()、.commit()など）を認識し、より良いサポート（補完など）を提供できます。型の宣言は、実際のオブジェクトには影響しません。

FastAPIのパスオペレーションを作成します¶

最後に、標準的なFastAPIのパスオペレーションコードを以下に示します。

Python 3.9+Python 3.8+

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from typing import List

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

依存関係でyieldを使用して、各リクエストの前にデータベースセッションを作成し、後でクローズしています。

そして、そのセッションを直接取得するために、パスオペレーション関数で必要な依存関係を作成できます。

これにより、パスオペレーション関数内から直接crud.get_userを呼び出して、そのセッションを使用できます。

豆知識

返される値は、SQLAlchemy モデルまたは SQLAlchemy モデルのリストであることに注意してください。

しかし、すべてのパスオペレーションは、orm_modeを使用する Pydantic モデル / スキーマでresponse_modelを持ち、Pydantic モデルに宣言されたデータがそれらから抽出されてクライアントに返されます。通常のフィルタリングと検証がすべて行われます

豆知識

また、List[schemas.Item]のような標準的な Python 型を持つresponse_modelsがあることに注意してください。

しかし、そのListの内容/パラメーターは、orm_modeを持つ Pydantic モデルなので、データは通常どおり取得されてクライアントに返され、問題はありません。

`def` vs `async def`について¶

ここでは、パスオペレーション関数と依存関係の中で SQLAlchemy コードを使用しており、それが外部データベースとの通信を行うことになります。

これにより、潜在的に「待機」が必要になる可能性があります。

しかし、SQLAlchemy は、次のようなもので使用されるawaitを直接使用するための互換性はありません。

user = await db.query(User).first()

...そして、代わりに以下のものを使用しています。

user = db.query(User).first()

Then we should declare the path operation functions and the dependency without async def, just with a normal def, as:

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    ...

情報

リレーショナルデータベースに非同期的に接続する必要がある場合は、非同期 SQL（リレーショナル）データベースを参照してください。

非常に技術的な詳細

興味があり、深い技術知識をお持ちの場合は、このasync def vs defがどのように処理されるかについて、非同期ドキュメントの非常に技術的な詳細を確認できます。

マイグレーション¶

SQLAlchemy を直接使用しており、FastAPIとの連携のためにプラグインは必要ないため、Alembicを使用してデータベースのマイグレーションを直接統合できます。

また、SQLAlchemy と SQLAlchemy モデルに関連するコードは、独立した別々のファイルに存在するため、FastAPI、Pydantic、またはその他のものをインストールすることなく、Alembic でマイグレーションを実行することもできます。

同じように、FastAPIに関係しないコードの他の部分で、同じ SQLAlchemy モデルとユーティリティを使用できます。

たとえば、Celery、RQ、またはARQを使用したバックグラウンドタスクワーカーの場合です

すべてのファイルの振り返り¶

my_super_projectという名前のディレクトリがあり、その中にsql_appという名前のサブディレクトリがあることを覚えておいてください。

sql_app は以下のファイルを持っているべきです。

sql_app/__init__.py:　空のファイル.
sql_app/database.py:

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

sql_app/models.py:

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

sql_app/schemas.py:

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


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


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

sql_app/crud.py:

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

sql_app/main.py:

Python 3.9+Python 3.8+

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from typing import List

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

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

確認してください¶

このコードをコピーしてそのまま使用できます。

情報

実際、ここで示されているコードはテストの一部です。このドキュメントのほとんどのコードと同じです。

次に、Uvicorn で実行できます。

$ uvicorn sql_app.main:app --reload

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

そしてブラウザでこのリンクを開いてください。 http://127.0.0.1:8000/docs.

そして、実際のデータベースからデータを読み込みながら、FastAPIアプリケーションとやり取りできるようになります。

Interact with the database directly¶

FastAPI とは別に、SQLite データベース（ファイル）を直接操作して、その内容をデバッグしたり、テーブル、列、レコードを追加したり、データを変更したりする場合は、DB Browser for SQLiteを使用できます。

以下のような表示になります。

SQLite ViewerやExtendsClassのようなオンラインの SQLite ブラウザを使用することもできます。

ミドルウェアを使用した代替 DB セッション¶

yieldを使用した依存関係を使用できない場合（たとえば、Python 3.7を使用しておらず、Python 3.6のバックポートをインストールできない場合）、同様の方法で「ミドルウェア」でセッションを設定できます。

「ミドルウェア」は基本的に、各リクエストに対して常に実行される関数であり、エンドポイント関数の実行前と実行後にいくつかのコードが実行されます。

Create a middleware¶

追加するミドルウェア（単なる関数）は、各リクエストに対して新しい SQLAlchemy のSessionLocalを作成し、リクエストに追加して、リクエストが完了したらクローズします。

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response


# Dependency
def get_db(request: Request):
    return request.state.db


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from typing import List

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response


# Dependency
def get_db(request: Request):
    return request.state.db


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

情報

SessionLocal()の作成とリクエストの処理をtryブロックに入れます。

そして、finallyブロックでクローズします。

これにより、リクエスト後にデータベースセッションが常にクローズされるようにします。リクエストの処理中に例外が発生した場合でもです。

`request.state`について¶

request.stateは各Requestオブジェクトのプロパティです。ここでは、この場合のデータベースセッションのように、リクエスト自体に添付された任意のオブジェクトを格納するために使用されます。 Starlette のRequest状態に関するドキュメントで詳細を確認できます。

この場合、これはリクエスト全体で単一のデータベースセッションが使用され、後で（ミドルウェアで）クローズされることを保証するのに役立ちます。

`yield`を使用した依存関係またはミドルウェア¶

ここでミドルウェアを追加することは、yieldを使用した依存関係が実行することと似ていますが、以下に示すようないくつかの違いがあります。

より多くのコードが必要で、少し複雑です。
ミドルウェアはasync関数である必要があります。
ネットワークを「待つ」必要があるコードが含まれている場合、そこでアプリケーションが「ブロック」され、パフォーマンスが少し低下する可能性があります。
SQLAlchemyの動作方法では、おそらくそれほど問題ではありません。
しかし、ミドルウェアに多くのI/O待機を含むコードを追加した場合、問題になる可能性があります。
ミドルウェアはすべてのリクエストに対して実行されます。
つまり、すべてのリクエストに対して接続が作成されます。
そのリクエストを処理するパスオペレーションが DB を必要としなかった場合でもです。

!!! tip 　"豆知識" ユースケースで十分な場合は、yieldを使用した依存関係を使用する方がおそらく良いでしょう。

情報

yieldを使用した依存関係は、最近FastAPIに追加されました。

このチュートリアルの以前のバージョンでは、ミドルウェアを使用した例のみが示されており、ミドルウェアを使用してデータベースセッションを管理しているアプリケーションがいくつか存在する可能性があります。

SQL (Relational) Databases¶

ORMs(オブジェクト関係マッピング)¶

ファイル構造¶

SQLAlchemyをインストールする¶

SQLAlchemy のパーツを作成する¶

SQLAlchemy のパーツをインポートする¶

SQLAlchemy 用のデータベース URL を作成する¶

SQLAlchemy engine を作成する¶

備考¶

SessionLocal クラスを作成する¶

Base クラスを作成する¶

Create the database models¶

Base クラスから SQLAlchemy モデルを作成する¶

モデルの属性/列を作成する¶

関係(relationship)の作成¶

Create the Pydantic models¶

初期の Pydantic モデル / スキーマを作成する¶

SQLAlchemy スタイルと Pydantic スタイル¶

読み取り/返却用の Pydantic モデル / スキーマを作成する¶

Pydantic の orm_mode を使用する¶

ORM モードに関する技術的な詳細¶

CRUD ユーティリティ¶

データの読み込み¶

データの作成¶

メインの FastAPI アプリ¶

データベーステーブルの作成¶

Alembic メモ¶

依存関係を作成します¶

FastAPIのパスオペレーションを作成します¶

def vs async defについて¶

マイグレーション¶

すべてのファイルの振り返り¶

確認してください¶

Interact with the database directly¶

ミドルウェアを使用した代替 DB セッション¶

Create a middleware¶

request.stateについて¶

yieldを使用した依存関係またはミドルウェア¶

`SQLAlchemy`をインストールする¶

SQLAlchemy `engine` を作成する¶

`SessionLocal` クラスを作成する¶

`Base` クラスを作成する¶

`Base` クラスから SQLAlchemy モデルを作成する¶

Pydantic の `orm_mode` を使用する¶

`def` vs `async def`について¶

`request.state`について¶

`yield`を使用した依存関係またはミドルウェア¶