Python fast api SQLとの連携方法についてまとめてみた

SQL (Relational) Databases

FastAPIはSQL（リレーショナル）データベースを使用する必要はありません。
しかし、任意のリレーショナルデータベースを使用することができます。
ここでは SQLAlchemy を使用した例を見てみましょう。
以下のような SQLAlchemy でサポートされているデータベースに簡単に適応させることができます。

• PostgreSQL
• MySQL
• SQLite
• Oracle
• Microsoft SQL Server, etc.
  今回の例では、SQLiteを使用しますが、これは単一のファイルを使用していることと、Pythonが統合的にサポートしているためです。そのため、この例をコピーしてそのまま実行することができます。

ORMs

• FastAPI は、任意のデータベースと任意のスタイルのライブラリを使用してデータベースと対話します。一般的なパターンは、「ORM」：「オブジェクトリレーショナルマッピング」ライブラリを使用することです。
• ORMには、コード内のオブジェクトとデータベーステーブル（「リレーション」）の間で変換（「マップ」）するツールがあります。
• ORMでは通常、SQLデータベースのテーブルを表すクラスを作成し、そのクラスの各属性は名前と型を持つカラムを表します。
• 例えば、クラス Pet は SQL テーブル pets を表すことができます。そして、そのクラスの各インスタンスオブジェクトはデータベース内の行を表します。
• 例えば、オブジェクト orion_cat (Pet のインスタンス) は、列の型を表す属性 orion_cat.type を持つことができます。そして、その属性の値は、例えば "cat "となります。これらのORMは、テーブルやエンティティ間の接続やリレーションを作成するツールも持っています。
• このようにして、orion_cat.ownerという属性を持つことができ、そのownerには、テーブルの所有者から取得したこのペットの所有者のデータが含まれます。つまり、orion_cat.owner.nameは、このペットの飼い主の名前(ownersテーブルのname列から)になります。これは "Arquilian "のような値を持つことができます。
• そして、ペットオブジェクトからアクセスしようとしたときに、対応するテーブルの所有者から情報を取得するためのすべての作業をORMが行います。
• ここでは SQLAlchemy ORM での作業方法を見てみましょう。

ファイル構成

この例では、my_super_projectという名前のディレクトリがあり、その中にsql_appというサブディレクトリがあり、以下のような構造になっているとします。

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

SQLAlchemy用のDBのURLを作る

• この例では、SQLiteデータベースに「接続」しています（SQLiteデータベースでファイルを開く）。
  -そのファイルは、同じディレクトリにある sql_app.db というファイルになります。最後の部分が./sql_app.dbになっているのはそのためです。

SQLAlchemyエンジンを作る

• まずはSQLAlchemyの「エンジン」を作成します。このエンジンは後に他の場所で使うことになります。
• 引数connect_args={"check_same_thread": False}はSQlite用の物です。

SessionLocalクラスの作成

• SessionLocalクラスの各インスタンスはデータベースセッションになります。クラス自体はまだデータベースセッションではありません。
• しかし、一度SessionLocalクラスのインスタンスを作成すると、このインスタンスが実際のデータベース・セッションになります。
• SQLAlchemyからインポートしているSessionと区別するためにSessionLocalと名付けました。後でSession（SQLAlchemyからインポートしたもの）を使用します。
• SessionLocalクラスを作成するには、関数sessionmakerを使用します。

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

DBモデルの作成

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

• SQLAlchemyでは、データベースと相互作用するこれらのクラスやインスタンスを指すために「モデル」という用語を使用しています。
• しかし、Pydanticはまた、データの検証、変換、ドキュメンテーションのクラスとインスタンスという異なるものを指すために「モデル」という用語を使用しています。
• tablename はSQLAlchemyでデータベースで使用されているテーブルの名前を指定します。

モデルの属性/カラムの作成

• ここで、すべてのモデル（クラス）属性を作成します。これらの属性はそれぞれ対応するデータベーステーブルのカラムを表します。
• デフォルト値としてSQLAlchemyのColumnを使用します。
• そして、データベースの型を定義するSQLAlchemyクラス「type」をInteger、String、Booleanとして引数に渡します。

relationshipの作成

• ここでリレーションを作成します。
• このために、SQLAlchemy ORMが提供するリレーションを使用します。
• これは、多かれ少なかれ、この1つに関連する他のテーブルからの値を含む "魔法の "属性になります。
• my_user.itemsのように、Userの属性アイテムにアクセスすると、Userテーブルのこのレコードを指す外部キーを持つ（アイテムテーブルからの）アイテムSQLAlchemyモデルのリストが表示されます。
• my_user.itemsにアクセスすると、SQLAlchemyは実際にアイテムテーブルのデータベースからアイテムをフェッチして、ここに入力します。
• そして、アイテム内の属性所有者にアクセスすると、ユーザー テーブルからユーザー SQLAlchemy モデルが含まれます。owner_id属性/カラムを外部キーで使用して、ユーザーテーブルからどのレコードを取得するかを知ることができます。

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, index=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, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

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

Pydanticモデルの作成

• SQLAlchemyモデルとPydanticモデルの混乱を避けるために、SQLAlchemyモデルのファイルmodels.pyとPydanticモデルのファイルschemas.pyを用意します。
• これらのPydanticモデルは多かれ少なかれ「スキーマ」（有効なデータの形）を定義します。したがって、これは両方を使用している間の混乱を避けるのに役立ちます。
• データの作成や読み込み中に共通の属性を持つように、ItemBaseとUserBaseのPydanticモデル（あるいは「スキーマ」とでも言いましょうか）を作成します。そして、それらを継承するItemCreateとUserCreateを作成し（だから同じ属性を持つことになる）、さらに作成に必要なデータ（属性）を追加します。つまり、ユーザーは作成時にパスワードも持つことになります。しかし、セキュリティのために、パスワードは他のPydanticモデルには存在しません、例えば、ユーザーを読み込む際にAPIから送信されることはありません。

SQLAlchemy style and Pydantic style

SQLAlchemyのモデルでは=を使用して属性を定義し、以下のように型をパラメータとしてColumnに渡すことに注意してください。
Pydantic モデルは : を使用して型を宣言していますが、新しい型注釈構文/型ヒントを使用しています。

読み込み/返却用のPydanticモデル/スキーマの作成

• ここで、データを読み込むとき、APIから返すときに使用するPydanticのモデル（スキーマ）を作成します。例えば、アイテムを作成する前は、アイテムに割り当てられたIDが何になるのかわかりませんが、読み込むとき（APIから返すとき）にはすでにIDがわかっています。
• 同様に、ユーザーを読み込むときに、アイテムにはこのユーザーに属するアイテムが含まれることを宣言できるようになりました。
• それらのアイテムのIDだけでなく、アイテムを読み込むためのPydanticモデルで定義したすべてのデータが含まれています。アイテムを読み込むためのPydanticモデルで定義したすべてのデータ。

orm_modeについて

• さて、読み込み、アイテム、ユーザーのPydanticモデルの中に、内部のConfigクラスを追加します。このConfigクラスはPydanticに設定を提供するために使用します。
• Configクラスの中で、ORM_MODE = Trueという属性を設定します。
• PydanticのORM_MODEは、データがdictではなくORMモデル（または属性を持つ他の任意のオブジェクト）であっても、Pydanticモデルにデータを読み込むように指示します。そして、これでPydanticのモデルはORMと互換性があり、パス操作でresponse_modelの引数に宣言するだけでOKです。データベースモデルを返すことができるようになり、そこからデータを読み込んでくれるようになります。

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

• SQLAlchemyをはじめとする多くの機能は、デフォルトでは "遅延読み込み "になっています。
• これは、例えば、データを含む属性にアクセスしない限り、データベースからリレーションシップのデータを取得しないことを意味します。例えば、属性項目にアクセスするとします。

current_user.items

• itemsにアクセスすると、SQLAlchemyはアイテムテーブルに行き、このユーザーのアイテムを取得しますが、それ以前は取得できません。orm_modeがなければ、Pydanticのモデルでそれらのリレーションシップを宣言していたとしてもパス操作からSQLAlchemyモデルを返しても、リレーションシップデータは含まれません。
• しかしORMモードでは、Pydantic自身が必要なデータを属性からアクセスしようとするので（ディクトを想定するのではなく）、返したい特定のデータを宣言することができ、ORMからでもそれを取りに行くことができるようになります。

from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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

CRUD

CRUDは、Create、Read、Update、Deleteから来ています。...この例では、作成と読み込みだけですが。

READ

• sqlalchemy.ORMからSessionをインポートしてください。これにより、dbパラメータの型を宣言することができ、関数の型チェックと補完がより簡単になります。モデル（SQLAlchemyモデル）とスキーマ（Pydanticモデル/スキーマ）をインポートします。ユーティリティ関数を作成します。
• パス操作関数とは独立して、データベースとのやりとり（ユーザーやアイテムを取得する）だけに特化した関数を作成することで、複数の部分での再利用が容易になり、また、そのためのユニットテストも追加することができます。

CREATE

• 次に、データを作成するユーティリティ関数を作成します。手順は以下の通りです。
  • データを使用してSQLAlchemyモデルのインスタンスを作成します。
  • そのインスタンス・オブジェクトをデータベース・セッションに追加します。
  • 変更をデータベースにコミットします（保存されます）。
  • インスタンスをリフレッシュします（生成されたIDなど、データベースからの新しいデータが含まれるように）。
• 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)

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

Main APP

Create the database tables

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

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 を使用します。そして、"migration"のためにもAlembicを使用します（それが主な仕事です）。
• migrationとは、SQLAlchemy モデルの構造を変更したり、新しい属性を追加したりする際に、その変更をデータベースに複製したり、新しいカラムや新しいテーブルを追加したりするために必要となる一連のステップのことです。
• FastAPIプロジェクト内のAlembicの例は、Project Generation - Templateからテンプレートで確認することができます。具体的には、ソースコードのalembicディレクトリにあります。

Dependencyを作る

• sql_app/databases.pyファイルで作成したSessionLocalクラスを使って依存関係を作成します。リクエストごとに独立したデータベースのセッション/接続(SessionLocal)を持ち、すべてのリクエストで同じセッションを使用し、リクエストが終了したら閉じます。そして、次のリクエストのために新しいセッションが作成されます。
• そのために、先に説明したように、yieldを使った新しい依存関係を作成します。
• この依存関係は、単一のリクエストで使用される新しいSQLAlchemy SessionLocalを作成し、リクエストが終了したらそれを閉じます。
• SessionLocal()の作成とリクエストの処理をtryブロックに入れています。そして、最後のブロックでそれを閉じます。
• このようにして、リクエストの後にデータベースセッションが常に閉じられるようにしています。たとえリクエストの処理中に例外が発生したとしても。
• しかし、(yield後の)終了コードから別の例外を発生させることはできません。
• そして、パス操作関数で依存関係を使用するときは、SQLAlchemyから直接インポートしたSession型で宣言します。これにより、パス操作関数内でのエディタのサポートが向上し、エディタはdbパラメータがSession型であることを知ることができます。
• パラメータ db は実際には SessionLocal 型ですが、このクラス (sessionmaker() で作成された) は SQLAlchemy セッションの「プロキシ」なので、エディタはどのようなメソッドが提供されているかを知りません。
• しかし、型を Session と宣言することで、エディタは利用可能なメソッド (.add(), .query(), .commit() など) を知ることができ、より良いサポート (補完など) を提供することができるようになりました。型宣言は実際のオブジェクトには影響しません。
• さて、最後に標準のFastAPIのパス操作コードです。
• yieldで依存関係にある各リクエストの前にデータベースのセッションを作成し、その後に閉じています。そして、そのセッションを直接取得するために、パス操作関数の中に必要な依存関係を作成しています。
• これで、パス操作関数の中から直接crud.get_userを呼び出して、そのセッションを利用することができます。

同期関数と非同期関数について

• ここでは、パス操作関数と依存関係の中でSQLAlchemyコードを使用しています。
• これは潜在的に「待ち時間」を必要とする可能性があります。
• しかし、SQLAlchemyには await を直接使用するための互換性がないので、以下のようなものを使用します。

Migrations

• SQLAlchemy を直接使用しており、FastAPI で動作するためのプラグインの種類を必要としないため、データベースの移行を Alembic で直接統合することができます。
• また、SQLAlchemyとSQLAlchemyモデルに関連するコードが別個の独立したファイルに存在するため、FastAPIやPydanticなどをインストールすることなくAlembicで移行を実行することさえ可能になります。