SQL (リレーショナル) データベース

情報

これらのドキュメントはまもなく更新されます。 🎉

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

新しいドキュメントには、Pydantic v2が含まれ、SQLModel（これもSQLAlchemyに基づいています）がPydantic v2を使用するように更新されたら使用します。

FastAPI では、SQL（リレーショナル）データベースの使用は必須ではありません。

しかし、好きなリレーショナルデータベースを使用できます。

ここでは、SQLAlchemy を使用した例を見ていきます。

PostgreSQL のように、SQLAlchemy でサポートされている任意のデータベースに簡単に適合させることができます。

• PostgreSQL
• MySQL
• SQLite
• Oracle
• Microsoft SQL Server など。

この例では、1つのファイルを使用し、Pythonに統合サポートがあるため、SQLiteを使用します。したがって、この例をコピーしてそのまま実行できます。

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

ヒント

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

注意

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

FastAPI固有のコードは、いつもどおり最小限です。

ORM

FastAPIは、あらゆるデータベース、およびデータベースと通信するためのあらゆるスタイルのライブラリで動作します。

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

ORMには、コード内のオブジェクトとデータベースのテーブル（「関係」）の間を変換（「マッピング」）するツールがあります。

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

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

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

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

これらのORMには、テーブルやエンティティ間の接続や関係を作成するツールもあります。

このようにして、属性orion_cat.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は単なる空のファイルですが、sql_appとすべてのモジュール（Pythonファイル）がパッケージであることをPythonに伝えます。

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

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」を作成することです。

この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つのスレッドのみがSQLiteと通信することを許可します。

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

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

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

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()

データベースモデルを作成する

次に、ファイル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")

関係を作成する

次に、関係を作成します。

このために、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")

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

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

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

Pydanticモデルを作成する

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

ヒント

SQLAlchemyのモデルとPydanticのモデルの混同を避けるために、SQLAlchemyモデルを含むmodels.pyファイルと、Pydanticモデルを含むschemas.pyファイルを用意します。

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

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

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

データの作成または読み取り中に共通の属性を持つように、ItemBaseおよびUserBase Pydanticモデル（または「スキーマ」としましょう）を作成します。

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

そのため、ユーザーは作成時にpasswordも持つことになります。

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

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は、Pydanticモデルに、dictではなくORMモデル（または属性を持つ任意のオブジェクト）であってもデータを読み取るように指示します。

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

id = data["id"]

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

id = data.id

これにより、PydanticモデルはORMと互換性があり、パス操作のresponse_model引数で宣言するだけで済みます。

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

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

SQLAlchemyなど多くは、デフォルトで「遅延読み込み」です。

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

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

current_user.items

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

orm_modeがない場合、*パス操作*からSQLAlchemyモデルを返しても、関係データは含まれません。

Pydanticモデルでそれらの関係を宣言した場合でもです。

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

CRUDユーティリティ

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

このファイルには、データベース内のデータと対話するための再利用可能な関数があります。

CRUDは、**C**reate（作成）、**R**ead（読み取り）、**U**pdate（更新）、**D**elete（削除）に由来します。

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

データの読み取り

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

models（SQLAlchemyモデル）とschemas（Pydanticの*モデル*/スキーマ）をインポートします。

以下のためのユーティリティ関数を作成します。

• 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

ヒント

*パス操作関数*とは独立して、データベースとの対話（ユーザーまたはアイテムの取得）にのみ専念する関数を作成することで、複数の場所でより簡単に再利用でき、それらのユニットテストを追加することもできます。

データの作成

次に、データを作成するためのユーティリティ関数を作成します。

手順は次のとおりです。

• データを使用してSQLAlchemyモデルの*インスタンス*を作成します。
• そのインスタンスオブジェクトをデータベースセッションにaddします。
• 変更をデータベースにcommitします（保存されるように）。
• インスタンスをrefreshします（生成された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にそれぞれ渡して、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を「マイグレーション」（それが主な仕事です）にも使用します。

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

FastAPIプロジェクトでのAlembicの例は、フルスタックFastAPIテンプレートにあります。具体的には、ソースコードのalembicディレクトリにあります。

依存関係の作成

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

リクエストごとに独立したデータベースセッション/接続（SessionLocal）を持ち、リクエスト全体で同じセッションを使用し、リクエストが完了したらそれを閉じる必要があります。

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

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

依存関係は、単一のリクエストで使用される新しいSQLAlchemySessionLocalを作成し、リクエストが完了したらそれを閉じます。

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を使用した依存関係で詳しく見てください。

そして、*パス操作関数*で依存関係を使用する場合、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とasync defについて

ここでは、*パス操作関数*と依存関係の中でSQLAlchemyコードを使用しており、それが今度は外部データベースと通信します。

それには「待機」が必要になる可能性があります。

ただし、SQLAlchemyは、次のようなawaitを直接使用するための互換性がないため

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

...代わりに次を使用しています。

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

次に、次のように、*パス操作関数*と依存関係をasync defなしで、通常のdefだけで宣言する必要があります。

@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対defがどのように処理されるかの非常に技術的な詳細をAsyncドキュメントで確認できます。

マイグレーション

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

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

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

例えば、Celery、RQ、またはARQを使用したバックグラウンドタスクワーカーなどです。

すべてのファイルを確認する

sql_appというサブディレクトリを含むmy_super_projectという名前のディレクトリが必要です。

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アプリケーションと対話できるようになります。

データベースと直接対話する

FastAPIとは独立して、SQLiteデータベース（ファイル）を直接調べ、その内容をデバッグしたり、テーブル、カラム、レコードを追加したり、データを変更したりしたい場合は、DB Browser for SQLiteを使用できます。

これは次のようになります。

また、SQLite ViewerやExtendsClassのようなオンラインSQLiteブラウザも使用できます。

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

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

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

ミドルウェアを作成する

追加するミドルウェア（単なる関数）は、リクエストごとに新しい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を必要としない場合でも。

ヒント

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

情報

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

このチュートリアルの以前のバージョンには、ミドルウェアを使用した例しかなく、データベースセッション管理にミドルウェアを使用しているアプリケーションがいくつかあるでしょう。