Chapter02 FastAPIで重要なPython文法の復讐

シンタックスシュガー

プログラミング言語において、読み書きのしやすさのために導入される書き方

# 例：
@wrapper(example)
# 同じ
wrapper(expample)()

Chapter04 Dockerイメージの作成

docker-compose.yaml

version: '3'
services:
  demo-app:
    build: .
    volumes:
      - .dockervenv:/src/.venv
      - .:/src
    ports:
      - 8000:8000
    environment:
      - WATCHFILES_FORCE_POLLING=true

.venv

Pythonの仮想環境に関連するファイルが格納されるディレクトリ。仮想環境は、Pythonのプロジェクトがその依存関係をプロジェクトごとに分離して管理できるようにするためのツールです。これにより、異なるプロジェクトで異なるバージョンのライブラリを使用することが可能になり、システム全体のPython環境を変更することなく、プロジェクトの依存関係を管理できます。

• 公式ドキュメント

WATCHFILES_FORCE_POLLING

Uvicornで使用される環境変数。
WSLを通じてUvicornを使用する際に、ファイル変更で自動的にリロードされるようにするために設定する。

• 公式ドキュメント

Dockerfile

Dockerfile

FROM python:3.11-buster

# pythonの出力表示をDocker用に調整
ENV PYTHONUNBUFFERED=1

WORKDIR /src

RUN pip install poetry

COPY pyproject.toml* poetry.lock* ./

RUN poetry config virtualenvs.in-project true
RUN if [ -f pyproject.toml ]; then poetry install --no-root; fi

ENTRYPOINT [ "poetry", "run", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--reload" ]

PYTHONUNBUFFERED

Pythonで使用される環境変数。
この環境変数を設定すると、標準出力と標準エラー出力が即座にコンソールに出力されるようになる。

出力の即時性は向上しますが、出力の量が多い場合はパフォーマンスに影響を与える可能性がある。また、出力の順序が保証されるわけではないため、複数の出力ソースがある場合は注意が必要。

• 公式ドキュメント

poetry config virtualenvs.in-project true

venvのファイルをプロジェクトディレクトリの下に置くための設定コマンド

poetry install --no-root;

--no-rootオプションを指定することで、プロジェクト自体（つまり、現在作業しているプロジェクトのルートパッケージ）はインストールされず、依存関係のみがインストールされる。

poetry run uvicorn api.main:app --host 0.0.0.0 --reload

api.main:app: uvicorn に渡される引数で、api.main モジュール（Pythonファイル）内のappインスタンス（ASGIアプリケーション）を指す。このappが起動するWebアプリケーション。

--host 0.0.0.0: Webサーバーがすべてのネットワークインターフェースでリッスンするように指定します。これにより、コンテナが稼働しているネットワーク上の任意のアドレスからアクセス可能になります。

--reload: コードに変更があった場合にサーバーを自動的に再起動するためのオプション。開発中に有用ですが、本番環境での使用は推奨されない。

• FastAPIでホットリロードを有効にする方法(2パターン)
• 公式ドキュメント

Chapter05 FastAPIのインストール

PoetryによるPython環境のセットアップ

$ docker compose run \
> --entrypoint "poetry init \
> --name demo-app \
> --dependency fastapi \
> --dependency uvicorn[standard]" \
> demo-app

Poetryについて

Pythonのパッケージマネージャ
pipの上位互換...?
JavaScriptのnpmやyarnと同じ

• Poetryをサクッと使い始めてみる
• 公式ドキュメント

poetry init

既にソースコードを書き始めてしまった場合、あるいは既にpipなどで管理しているPythonのプロジェクトでpoetryを使い始める場合には poetry initを使用。
まっさらな状態から新たにプロジェクトを作る場合はpoetry newを使用。

--dependencyオプション

プロジェクトが依存するライブラリ（依存関係）を指定するために使用。

Chapter09 スキーマ(Schemas)-レスポンス

api/schemas/task.py

import datetime
from pydantic import BaseModel, Field


class TaskBase(BaseModel):
	title: str | None = Field(None, example="クリーニングを取りに行く")
	due_date: datetime.date | None = Field(None, example="2024-12-01")


class TaskCreate(TaskBase):
	pass


class TaskCreateResponse(TaskCreate):
	id: int

	class Config:
		orm_mode = True


class Task(TaskBase):
	id: int
	done: bool = Field(False, description="完了フラグ")

	class Config:
		orm_mode = True

pydantic

Pythonでデータのバリデーション（検証）と設定管理を簡単に行うためのライブラリです。
データのバリデーションルールを明確にし、不正確なデータや型不一致を事前に防ぐ。

• Pydantic 入門
• 【Python開発に欠かせない！】pydanticの使い方とメリット
• 公式ドキュメント

Fieldモデル

モデルのフィールドをカスタマイズしてメタデータを追加するために使用

• 公式ドキュメント

BaseModel

スキーマモデルであることを表すクラス。スキーマを定義したいクラスはこれを継承する。

• 公式ドキュメント

api/routers/task.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

import api.schemas.task as task_schema

import api.cruds.task as task_crud
from api.db import get_db


router = APIRouter()


@router.get("/tasks", response_model=list[task_schema.Task])
async def list_tasks(db: AsyncSession = Depends(get_db)):
	return await task_crud.get_tasks_with_done(db)

router.get("/tasks", response_model=list[task_schema.Task])

• 公式ドキュメント

Chapter10 スキーマ(Schemas)-リクエスト

api/schemas/task.py

class TaskCreateResponse(TaskCreate):
	id: int

	class Config:
		orm_mode = True

orm_mode

Pydanticのconfig。
ORM(今回はSQLAlchemy)のモデルオブジェクトをPydanticのレスポンスオブジェクトに変換する。

• 【FastAPI+SQLAlchemy】orm_mode = Trueの設定メモ
• 公式ドキュメント

Chapter11 MySQLコンテナの立ち上げ

docker-compose.yaml

  db:
    image: mysql:8.0
    platform: linux/x86_64
    environment:
      MYSQL_ALLOW_EMPTY_PASSWORD: 'yes' # rootアカウントをパスワードなしで作成
      MYSQL_DATABASE: 'demo' # 初期データベースとしてdemoを設定
      TZ: 'Asia/Tokyo' # タイムゾーンを日本時間に設定
    volumes:
      - mysql_data:/var/lib/mysql
    command: --default-authentication-plugin=mysql_native_password # MySQL8.0ではデフォルトが"caching_sha2_password"で、ドライバが非対応のため変更
    ports:
      - 33306:3306 # ホストマシンのポート33306を、docker内のポート3306に接続する

--default-authentication-plugin=mysql_native_password

DockerコンテナでMySQL8.0イメージを使用する際に、MySQLサーバーのデフォルト認証プラグインをmysql_native_password に設定するためのもの。

• DockerでMySQL8の環境構築でハマった話

$ docker compose exec demo-app poetry add sqlalchemy pymysql

poetry add

開発途中で必要に応じて依存パッケージを追加する際に使用

• 公式ドキュメント

sqlalchemy

Pythonの中でよく利用されているORM。
ORMとは，Object Relational Mapperのことで，簡単に説明すると，テーブルとクラスを1対1に対応させて，そのクラスのメソッド経由でデータを取得したり，変更したりできるようにする。

• SQLAlchemyの基本的な使い方
• 公式ドキュメント

pymysql

PyMySQLは、Pythonプログラミング言語でMySQLデータベースに接続し、データを操作するための純粋なPython MySQLクライアントライブラリ。

• 公式ドキュメント

api/db.py

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


DB_URL = "mysql+pymysql://root@db:3306/demo?charset=utf8"


db_engine = create_engine(DB_URL, echo=True)
db_session = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)


Base = declarative_base()


def get_db():
	with db_session() as session:
		yield session

create_engine

SQLAlchemyを使用して同期的なデータベースエンジンを作成する。
SQLAlchemyのコア機能の一部であり、アプリケーションとデータベース間の通信を管理するエンジンのインスタンスを生成する。このエンジンは、データベースとのすべての通信の基礎。

• 公式ドキュメント

sessionmaker

SQLAlchemyを使用してセッションファクトリを作成する。
sessionmaker関数は、データベースセッションを生成するためのファクトリ（生成器）を設定する。このファクトリを使用して生成されたセッションは、データベースとのすべての通信を管理する。

• autocommit=False: トランザクションは自動的にコミットされない。明示的にcommit()を呼び出す必要がある。
• autoflush=False: セッションに追加または変更されたオブジェクトは、クエリの実行時に自動的にフラッシュ（データベースに書き込まれる前の準備状態）されない。
• bind=db_engine: データベースエンジンを指定する。

• 公式ドキュメント

declarative_base

モデルの基底となるクラスを生成する。モデルとして扱うクラスはこの関数によって生成したクラスを継承する。

• 公式ドキュメント

api/models/task.py

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

from api.db import Base


class Task(Base):
	__tablename__ = "tasks"

	id = Column(Integer, primary_key=True)
	title = Column(String(1024))
	due_date = Column(Date)

	done = relationship("Done", back_populates="task", cascade="delete")


class Done(Base):
	__tablename__ = "dones"

	id = Column(Integer, ForeignKey("tasks.id"), primary_key=True)

	task = relationship("Task", back_populates="done")

tablename

モデルクラスがデータベース内のどのテーブルにマッピングされるかを指定するために使用される。この属性を使って、クラスが操作するデータベーステーブルの名前を定義する。これにより、クラスのインスタンス（オブジェクト）とデータベーステーブルの行が関連付けられる。

• 公式ドキュメント

relationship

モデル間の関連を定義し、双方向の関連やカスケード削除などのオプションを提供する。

• 公式ドキュメント

$ docker compose exec demo-app poetry run python -m api.migrate_db

-m

モジュールを指定して実行
-mの後に続く指定されたモジュールをsys.pathから探し、それを__main__のスクリプトとして実行する。

Chapter12 DB操作(CRUDS)

api/cruds/task.py

from sqlalchemy import select
from sqlalchemy.engine import Result
from sqlalchemy.ext.asyncio import AsyncSession

import api.models.task as task_model
import api.schemas.task as task_schema


async def create_task(db: AsyncSession, task_create: task_schema.TaskCreate) -> task_model.Task:
	task = task_model.Task(**task_create.dict())
	db.add(task)
	await db.commit()
	await db.refresh(task)
	return task

###　**task_create.dict()

pydanticのBaseModelの関数で、辞書に変換をしてくれる。

• 公式ドキュメント

api/routers/task.py

from fastapi import APIRouter, Depends

@router.post("/tasks", response_model=task_schema.TaskCreateResponse)
async def create_task(task_body: task_schema.TaskCreate, db: Session = Depends(get_db)):
	return task_crud.create_task(db, task_body)

Depends

• [FastAPI] 依存性注入って結局なんで使うんじゃい？
• 公式ドキュメント

api/routers/done.py

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

import api.schemas.done as done_schema
import api.cruds.done as done_crud
from api.db import get_db

router = APIRouter()


@router.put("/tasks/{task_id}/done", response_model=done_schema.DoneResponse)
async def mark_task_as_done(task_id: int, db: Session = Depends(get_db)):
	done = task_crud.get_done(db, task_id=task_id)
	if done is not None:
		raise HTTPException(status_code=400, detail="Done already exists")
	
	return done_crud.create_done(db, task_id)


@router.delete("/tasks/{task_id}/done", response_model=None)
async def unmark_task_as_done(task_id: int, db: Session = Depends(get_db)):
	done = task_crud.get_done(db, task_id=task_id)
	if done is None:
		raise HTTPException(status_code=404, detail="Done not found")
	
	return done_crud.delete_done(db, original=done)

HTTPException

FastAPI フレームワークにおいて、HTTPリクエストの処理中に特定のHTTPステータスコードとメッセージでクライアントにエラーを返すために使用される例外クラス。

• [Python]FastAPI エラーハンドリング方法 メモ
• 公式ドキュメント

Chapter14 テストユニット

$ docker compose exec demo-app poetry add -G dev pytest-asyncio aiosqlite httpx

-G

追加する依存関係を特定のグループ（この場合は dev）に指定。

pytest-asyncio

pytestテストフレームワークで非同期コードのテストをサポートするためのパッケージ。

• pytestで非同期関数をテストする
• 公式ドキュメント

httpx

非同期HTTPクライアントであり、requests ライブラリの非同期版とも言えます。HTTP/1.1とHTTP/2の両方をサポートし、非同期プログラムでの外部API呼び出しに使用される。

• HTTPXとasyncioを利用したPythonの非同期HTTPリクエスト
• 公式ドキュメント

starlette

同期Pythonフレームワークの構築に焦点を当てた、軽量なASGIフレームワーク。FastAPIの基盤としても知られており、非同期Webアプリケーションやマイクロサービスの開発に適している。Pythonのasyncioライブラリに基づいて構築されている。

• 公式ドキュメント

追加課題 Postgresへの移行

psycopg2

Pythonプログラミング言語用のPostgreSQLデータベースアダプター。
同期的な操作のみをサポートしている。非同期IOを使用したい場合は、asyncioに対応したasyncpgのような他のライブラリを検討する必要がある。

• psycopg2 でよくやる操作まとめ
• 公式ドキュメント

asyncpg

Pythonの非同期IOフレームワークであるasyncioを用いて、PostgreSQLデータベースに対する高速な非同期アクセスを提供するライブラリ。

• asyncpg でよくやる操作まとめ
• 公式ドキュメント