はじめに

本記事はFastAPI初心者が実装したお話を記載しています。

メインの参考書は以下です。（要約ページではないのでご注意を！）

追加で業務内での利用内容を含んでいるため、上記参考書通りの記載とは異なることをご了承下さい。
またDockerによる環境構築も行っていますが、今回の記事にはそちらは含まれません。あくまでFastAPIについてをまとめたものとなります。

FastAPIとは

FastAPIはPythonで利用される、最近？人気の高速Webフレームワークです。
特徴は、以下のように挙げられています。

高速。公式には、Python フレームワークの中で最も高速なものの 1 つとも
リクエスト・レスポンスのスキーマ定義を明示的に定義、型安全の開発が可能
Swagger UIが自動生成、利用ができる
非同期処理ができる

FastAPIは、非同期Webアプリケーションを開発および実行するためのインターフェース定義であるASGIをPythonアプリケーション特化であるUvicorn(ユビコーン)と言うサーバーを利用しています。

他にも、代表的なPythonのフレームワークには、Django、Flask、Bottleなど複数存在しています。

sqlalchemyとは

今回作成するFastAPIではMySQLとの接続するためにSQLalchemyを利用します。
SQLalchemyとは、PythonのORM（Object-Relational Mappingの略）で、データベースとPythonのオブジェクトの関連付けを行うことができます。
ORMを利用することにより、SQL文を直接書かずにPython内で完結できたり、さまざまなDB独自のSQL文に囚われることなく実行することが可能です。

説明もさておき、実際に作成していきます。

Routerの作成

Routerではパス定義をします。
正格にはパスオペレーション関数、つまり「パス」と「オペレーション」を定義します。

今回はAPIの設計指標の1つであるRESTAPIに順次て構成してきます。
作成する「パス」と「オペレーション」は「エンドポイント」と「HTTPメソッド（GETやPOST）」と置き換えて考えることができます。

今回作成するAPIの仕様は以下とします。

エンドポイント	HTTPメソッド	パスパラメータ	リクエスト	レスポンス	概要
`/tasks`	GET	-	-	{id:str,task:str,doneFlag:bool}	タスクの情報を取得する
`/tasks`	POST	-	{task:str,doneFlag:bool}	-	タスクの情報を登録する
`/tasks`	PUT	{id}	{task:str,doneFlag:bool}	-	一意のIDのタスクの情報を更新する
`/tasks`	DELETE	{id}	-	-	一意のIDのタスクの情報を物理削除する

RESTAPの設計思想上、「tasks」に関するAPIのため同じエンドポイントで異なるHTTPメソッドにより呼び分けられていることがわかります。

パスパラメータとクエリパラメータの違いについて

今回はすべてパスパラメータを利用していますが、パラメータを渡す方法としてクエリパラメータも存在しています。
簡単に言えば、パスパラメータは？の前にあり、属性やグループを示すもの。クエリパラメータは？の後にあり、操作に必要なIDなどを示すものです。
https://example/{pathpara}/tasks?querypara1=hoge&querypara2=fuga

ファイル構成で表すとこんな感じ。パスパラメータはドキュメントになる、クエリパラメータは変数になるイメージです。

example/
 ├── pathpara1
 └── pathpara2
      ├── tasks1.exe -- {querypara1=hoge, querypara2=fuga}
      ├── tasks2.exe -- {querypara1=hoge, querypara2=fuga}
      └── tasks3.exe -- {querypara1=hoge, querypara2=fuga}

より詳細はこちら：[RESTful API]パスパラメータ、クエリパラメータ、リクエストボディの違いと設計

実際のコードはこちら

routers>tasks.py

from typing import List

from fastapi import APIRouter

from src.models.base import HTTPError
from src.models.tasks import TasksRequest, TasksResponse
from src.repositories.tasks import TasksRepositoryDep

router = APIRouter(tags=["tasks"])


@router.get(
    "/tasks",
    response_model=List[TasksResponse],
    responses={
        403: {"model": HTTPError},
    },
)
def get_tasks(
    task_repo: TasksRepositoryDep,
    offset: int = 0,
    limit: int = 10,
):
    return task_repo.select_tasks(offset, limit)


@router.post(
    "/tasks",
    response_model=TasksResponse,
    responses={
        403: {"model": HTTPError},
    },
)
def post_tasks(
    body: TasksRequest,
    task_repo: TasksRepositoryDep,
):
    return task_repo.insert_task(body)


@router.put(
    "/tasks/{id}",
    response_model=TasksResponse,
    responses={
        403: {"model": HTTPError},
        404: {"model": HTTPError},
    },
)
def put_tasks(
    id: str,
    body: TasksRequest,
    task_repo: TasksRepositoryDep,
):
    return task_repo.update_task(id, body)


@router.delete(
    "/tasks/{id}",
    responses={
        403: {"model": HTTPError},
        404: {"model": HTTPError},
    },
)
def delete_tasks(
    id: str,
    task_repo: TasksRepositoryDep,
):
    task_repo.delete_task(id)

※レスポンスやリクエストの宣言は以下Shemaの作成で説明
※各関数から呼び出し内容は以下CRUDsの作成で説明
※CRUDs先のクラス呼び出し方法については以下DIで説明

また、Router作成のみではSwaggerには表示されません。作成したRouterインスタンスをFastAPIインスタンスに取り込む必要があります

main.py

from fastapi import FastAPI
from .routers import tasks,

tags_metadata = [
    {
        "name": "tasks",
        "description": "DEMOAPI",
    },
]
app = FastAPI(openapi_tags=tags_metadata)
app.include_router(tasks.router)

実際にローカルホスト＋/docs(例http://localhost:8000/docs)へアクセスしてみると、、？

このように作成した4つのAPIが簡単にわかりやすく表示されます！便利〜

Modelの作成

今回作成するテーブル仕様は以下とします。

カラム名	概要	データ型	デフォルト	桁数	PK	Null	備考
id	タスクID	UUID	UUID	-	○	False
task	タスク名	str	-	200	-	False
doneFlag	完了フラグ	bool	False	-	-	False
created_at	登録日付	datetime	現在時刻	-	-	False
updated_at	更新日付	datetime	現在時刻	-	-	False	更新あり

Modelにはテーブルの1つ1つのカラムをColumとして記載していきます。
カラムの設定として、第一引数にはそのカラムの型定義を記載します。

SQLalchemyでの型定義は以下公式を参照し、型宣言を行なってください。

参照SQLalchemy公式　ジェネリック型

第二引数以降はカラム設定を記載していきます。
よく利用する設定値はこのようにSQLAlchemyで記載できます。

制約	SQLAlchemy	説明
PK (Primary Key)	`primary_key=True`	主キー。テーブル内で一意にレコードを識別するために使用。（Unique＋index＋notnullが付与）
FK (Foreign Key)	`ForeignKey('table.column')`	外部キー。別のテーブルのカラムを参照し、リレーションを作成。
index	`index=True`	インデックスを作成。クエリパフォーマンスを向上。
NOT NULL	`nullable=False`	カラムに`NULL`値を許可しない。
Unique	`unique=True`	ユニーク制約。カラムの値が他のレコードと重複しないようにする。
Default	`default=value`	INSERTデフォルト値。レコードが作成される際に、指定されたデフォルト値を利用。
server_default	`server_default=value`	DDLデフォルト値。テーブルが作成される際に、指定されたデフォルト値を利用。
On Update	`onupdate=func.now()`	レコードが更新されるたびに、指定した値や関数でカラムを自動更新する。
Comment	`comment="description"`	カラムに対してコメントを追加できます。テーブル定義を明確にするためのメタデータとして使用。

こちらの構成をsqlalchemyで記載していくと、、？

model>tasks.py

from datetime import datetime
from uuid import uuid4

from sqlalchemy import Boolean, Column, DateTime, Unicode
from sqlalchemy_utils import UUIDType

# マッピングレジストリを作成
mapper_registry = registry()
BaseOrmModel = mapper_registry.generate_base()

# タイムゾーン: 日本時間
TZ_LOCAL = datetime.timezone(datetime.timedelta(hours=9))

# DBのモデル
class TasksOrm(BaseOrmModel):
    __tablename__ = "tasks"

    id = Column(UUIDType(binary=False), primary_key=True, default=uuid4)
    task = Column(Unicode(200), nullable=False)
    doneFlag = Column(Boolean, default=False,nullable=False)
    created_at = Column(
        DateTime, default=lambda: datetime.now(TZ_LOCAL), nullable=False
    )
    updated_at = Column(
        DateTime,
        default=lambda: datetime.now(TZ_LOCAL),
        onupdate=lambda: datetime.now(TZ_LOCAL),
        nullable=False,
    )

defaultとserver_defaultの違いについて

defaultはDBのカラムをInsertする際のデフォルト値。
server_defaultはDBのテーブルをCreateする際のデフォルト値。

なぜserver_defaultが重要になるかというと、すでに存在しているテーブルに対しNOT NULLなカラムを追加しようとする際、既存のデータに対し何を入れれば良いかわからないとマイグレーションエラーとなる。（default値は採用してくれない。）
そのため、defaultとは別に、server_defaultの宣言が必要になる。ということである。カラム追加時は既存データや既存カラムへの配慮を忘れずにしよう！

参照：SQLAlchemy/カラムのデフォルト値

Shemaの作成

APIのレスポンンスの型やリクエストの型などを定義します。
DB構成をもとに、受け渡し内容を定義します。

実際のコードはこちら

shema>tasks.py

from pydantic import BaseModel
from datetime import datetime
from uuid import UUID

# DBのモデルの型を定義
class Tasks(TasksOrm):
    id: UUID
    task: str
    deleted: bool
    doneFlag: float
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True
        orm_model = True

# APIレスポンスのモデル
class TasksResponse(BaseModel):
    id: UUID
    task: str
    doneFlag: bool
    created_at: datetime
    updated_at: datetime

# APIリクエストのモデル
class TaskRequest(BaseModel):
    task: str
    doneFlag: bool

ここで定義した型にそってAPIの受け渡しを行います。

Alembicとは

SQLalchemyで作成したDBモデルをもとにDBマイグレーション（DBの作成やアップグレード、ダウングレードを行うこと）をします。
ここで利用するのは、Pythonのマイグレーションツールであるalembicというものです。

細かいセットアップは省略しますが、先ほど作成したDBモデルをもとに自動生成で作成されたスクリプトはこのようになります。

alembic>script.py

from alembic import op
import sqlalchemy as sa
import sqlalchemy_utils as sau

# revision identifiers, used by Alembic.
revision = 'XXXX'
down_revision = 'XXXX'
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table('tasks',
    sa.Column('id', sau.UUIDType(binary=False), nullable=False),
    sa.Column('task', sa.Unicode(length=200), nullable=False),
    sa.Column('doneFlag', sa.Boolean(), nullable=False),
    sa.Column('created_at', sa.DateTime(), nullable=False),
    sa.Column('updated_at', sa.DateTime(), nullable=False),
    sa.PrimaryKeyConstraint('id')
    )

def downgrade() -> None:
    op.drop_table('tasks')

参照：DBマイグレーションツールのAlembicの使い方

CRUDsの作成

DB操作（C:Create作成、R:Read読み取り、U:Update更新、D:Delete削除）について定義をします。
APIが呼び出され、直接DBの操作をするクエリを記載します。
実際のコードはこちら

repositories>tasks.py

from typing import Annotated

from fastapi import Depends
from sqlalchemy import delete, insert, select, update
from sqlalchemy.orm import Session

from src.database import SessionDep
from src.models.tasks import TasksRequest, Tasks, TasksOrm


class TasksRepository:
    def __init__(self, session: Session):
        self.session = session
    
    # タスク 一覧参照
    def select_tasks(self, offset: int, limit: int):
        tasks = self.session.execute(
            select(TasksOrm)
            .order_by(TasksOrm.created_at)
            .offset(offset)
            .limit(limit)
        ).scalars()
        return [Tasks.model_validate(task) for task in tasks]
    
    # タスク 登録
    def insert_task(self, body: TasksRequest):
        task = self.session.execute(
            insert(TasksOrm).values(**body.model_dump()).returning(TasksOrm)
        ).scalar_one()
        return Tasks.model_validate(task)
    
    # タスク 更新
    def update_task(self, id: str, body: TasksRequest):
        task = self.session.execute(
            update(TasksOrm).where(TasksOrm.id == id).values(**body.model_dump()).returning(TasksOrm)
        ).scalar_one()
        return Tasks.model_validate(task)
    
    # タスク 削除
    def delete_task(self, id: str):
        task = self.session.execute(
            delete(TasksOrm).where(TasksOrm.id == id).returning(TasksOrm)
        ).scalar_one()
        return Tasks.model_validate(task)


def get_tasks_repository(session: SessionDep):
    return TasksRepository(session)


# Tasksリポジトリ取得の依存関係定義
TasksRepositoryDep = Annotated[TasksRepository, Depends(get_tasks_repository)]

DIについて

DIとはDependency Injectionの略で、依存性の注入と呼ばれます。
日本語だけでは理解難しいですが、（あくまで自分の理解上では）密な関数同士の結合をやめ、関数同士依存を受け渡す形で実装する。ということです。

今回の実装では、RouterClassからRepositoryClassを呼んでいます。
もし、DIを利用せずにRouterにてRepositoryをインスタンス化し必要関数を呼び出した場合、呼び出した関数はそこでしか利用できなければ、関数結果は呼び出した関数に依存するため、関数ごとのテストはできません。

また、今回呼び出しているRepositoryクラスには1クラスしかありませんが、複数のRepositoryを呼び出したり、初期化処理として利用できたりもします。

つまりDIを利用するメリットとしては、
・結合度の低下によるコンポーネント化の促進
・単体テストの効率化
が主となっています。（個人的利用感を含む）

参照：DI（依存性の注入）とは依存性を注入するということである、、？

参照：依存性注入（Dependency Injection: DI）について理解する

最終ディレクトリ構成

ルーター、モデル、スキーマ、CRUDsはDBごとのファイル（今回ではtasks）となり、一番外にメインファイルが存在する構成となります。

src/
 ├── routers（パス定義）
      └── tasks.py
 ├── model（DB構成）
      └── tasks.py
 ├── shema（型定義）
      └── tasks.py
 ├── repositories（DB操作）
      └── tasks.py
 └── main.py
 
alembic/
 └── versions.py（マイグレーションスクリプト）
    └── script.py
 └── env.py

まとめ

今回はFastAPIを用いてAPIを接続してみました。構成がシンプルな分、理解しやすく実装も難しくなくできました。他のフレームワークの違いも気になるところです。また記載していきまーす！