ChatGPT先生と挑むFastAPI #7｜実務向け設定と周辺知識

それではSTEP7を始めましょう！

STEP7では以下の内容を学習します。

環境によって変わる設定は、Pythonコードから分離します。

プログラムの処理
        ↓
Pythonコード

環境によって変わる設定
        ↓
環境変数

たとえばPython側では、DATABASE_URL = 環境変数から取得とだけしておけば、

開発環境
DATABASE_URL=開発DB

本番環境
DATABASE_URL=本番DB

と、同じPythonコードのまま外側から設定だけ変更できます。

さらにDBのパスワードやAPIキーなどはGitHubへ公開したくありません。そのような秘密情報をソースコードから分離する意味でも環境変数は重要です。

FastAPI固有の仕組みというより、これはWebアプリ全般で使う考え方ですね。

ここは混同しやすいところです。

環境変数と .env ファイルは同じものではありません。

本来の環境変数はOSやコンテナなどの実行環境側に設定される値です。

一方 .env は、以下のように環境変数として使いたい値をファイルへまとめておく仕組みです。

DATABASE_URL=postgresql+psycopg://...
APP_NAME=FastAPI Learning

特にローカル開発では、毎回手動で環境変数を設定するのは面倒なので .env がよく使われます。

そして今回FastAPI側では、Pydanticの pydantic-settings を使います。

公式ドキュメントでも、BaseSettings を使うことで環境変数を読み込み、Pydanticと同じように型変換・バリデーションできる構成が案内されています。たとえば環境変数は元々文字列ですが、int と定義しておけば設定値として整数へ変換できます。

つまり今回作っていく構成は、ざっくり言うと次のようになります。

.env / 環境変数
        ↓
pydantic-settings
        ↓
Settings
        ↓
FastAPIアプリから利用

DRFでいうと settings.py や環境変数を使った設定管理に近い役割ですが、FastAPI自体にはDjangoのような巨大な設定システムが組み込まれていません。必要な設定管理をPydanticなどと組み合わせて構成する、という違いがあります。

STEP 6時点では app/database.py で次のように直接環境変数を取得していました。

import os

DATABASE_URL = os.environ["DATABASE_URL"]

今回はこれを pydantic-settings の Settings クラス経由に変更します。

FastAPI公式でも、BaseSettings と SettingsConfigDict(env_file=".env") を使う形が案内されています。

app 配下に新しく config.py を作ってください。

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # DB接続先
    database_url: str

    # .envから設定値を読み込む
    # Settingsに定義されていない項目は無視する
    model_config = SettingsConfigDict(
        env_file=".env",
        extra="ignore",
    )


# アプリ内で共通して使用する設定
settings = Settings()

ここでポイントなのは、database_url: strとPython側では小文字で定義していても、.env 側のDATABASE_URL=...を読み込めることです。

また、

BaseSettings は環境変数から値を読み込み、型変換やバリデーションも行ってくれます。

また、今回の .env には POSTGRES_DB など、Settings に定義していない項目も存在します。

pydantic-settings はデフォルトでは、このような定義外の項目を .env から読み込むとエラーになるため、

extra="ignore"

を指定し、定義されていない項目を無視するようにしています。

これにより、FastAPIで必要な設定項目だけを Settings に定義できます。

現在のコードを次のように変更してください。

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

from app.config import settings


# 設定クラスからDB接続先を取得する
engine = create_engine(settings.database_url)

# DB操作で使用するSessionを生成する
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass

DB接続先について、変更前はOSの環境変数からDATABASE_URLを取得していましたが、変更後はapp/config.pyのsettings.database_urlを使うようにしています。

これで構造としては次のようになります。

.env
  ↓
Settings
  ↓
settings.database_url
  ↓
SQLAlchemy
  ↓
PostgreSQL

database.py 自身が「環境変数をどう取得するか」を知らなくてよくなるのが大きなポイントです。設定値の管理責任を config.py にまとめられます。

つまり、以下のような責任分担になります。

• config.py：設定値をどこから、どのように取得するかを管理する。
• database.py：取得済みの設定値を使ってDB接続を構築する。

Dev Container内のプロジェクトルートで、これまでどおりFastAPIを起動してください。

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

そのうえで、STEP 6で作った DBを使用する商品API をSwagger UIなどから実行してみてください。

既存の商品取得・登録などがこれまでどおり正常に動けば、

.env
↓
pydantic-settings
↓
SQLAlchemy
↓
PostgreSQL

という経路でDB接続情報を取得できています。

FastAPIでは、APIの処理を定義する際に、通常の def と async def の2種類を使用できます。

今回は、この違いと非同期処理の仕組みについて理解していきます。

例えば、APIで外部サービスにリクエストを送信する処理を考えます。

同期処理の場合、次のような流れになります。

外部APIへリクエスト
      ↓
応答待ち：3秒（待機中に次の処理へ進めない）
      ↓
応答を受信
      ↓
処理再開

同期処理では、基本的に応答が返ってくるまで、その実行の流れは待機します。

ただし、FastAPI全体がすべてのリクエストを処理できなくなるわけではありません。後述するように、FastAPIは同期APIをスレッドプールで実行できます。

一方、非同期処理の場合は次のような流れになります。

リクエストA：外部APIへアクセス
      ↓
Aは応答待ち（await）：3秒（Aの処理を一時停止し、ほかの処理を進められる）
リクエストBの処理を進める
      ↓
Aの応答が届く
      ↓
Aの処理を再開

非同期処理では、外部APIなどの応答を待っている間、別の処理を進められます。

このように、複数の処理を待ち時間を利用して切り替えながら進める仕組みを、並行処理（Concurrency）と呼びます。

※ 非同期処理は、CPUを複数使って処理を同時に実行する並列処理（Parallelism）とは異なります。

非同期処理を記述する場合は、次のように書きます。

@app.get("/example")
async def example():
    result = await fetch_data()
    return result

※ fetch_data() は非同期処理に対応した関数を想定した説明用の例です。

それぞれの役割は次のとおりです。

重要なのは、await は単純に処理を待つだけではなく、その待ち時間に別の処理を進められる点です。

FastAPIでは、イベントループという仕組みが、非同期処理の実行や再開を管理しています。

なお、async def と書くだけで、その関数内のすべての処理が自動的に非同期になるわけではありません。await できる非同期ライブラリなどを適切に使用する必要があります。（公式ドキュメント）

ここが今回の学習で特に重要です。

STEP 6では、次のようにDB接続を作成しました。

engine = create_engine(settings.database_url)

SessionLocal = sessionmaker(bind=engine)

この構成は、SQLAlchemyの同期処理用の構成です。

DB操作も同期的に行われるため、通常のAPI関数では次のように def を使用します。

@app.get("/products")
def get_products():
    # 同期的なDB操作
    ...

FastAPIは、通常の def で定義されたAPI関数をスレッドプールで実行するため、DB処理の待ち時間によってイベントループが直接ブロックされることを避けられます。

一方、async def に変更するだけでは、SQLAlchemyの同期DB操作が非同期になるわけではありません。

DB操作まで非同期化する場合は、create_async_engine() や AsyncSession などを使用する別の構成が必要です。

今回は既存のSQLAlchemyの構成を変更せず、同期処理と非同期処理の使い分けを学びます。

FastAPIだからといって、すべてのAPIを async def で実装する必要はありません。

• 非同期対応のライブラリを await して使用する場合：async def
• 同期的なDB操作など、ブロッキング処理を使用するAPIの場合：基本的に def

という使い分けができます。

今回は、同じ2秒の待機処理を同期・非同期それぞれで実装し、FastAPIがどのように実行するのか確認します。

確認するポイントは、待機時間ではなく処理が実行されるスレッドの違いです。

app/routers/execution_demo.py を新規作成してください。

import asyncio
import threading
import time

from fastapi import APIRouter

router = APIRouter(prefix="/execution-demo", tags=["Execution Demo"])


@router.get("/sync")
def sync_example():
    # 同期的に2秒間待機する
    time.sleep(2)

    return {
        "type": "sync",
        "thread": threading.current_thread().name,
    }


@router.get("/async")
async def async_example():
    # 非同期的に2秒間待機する
    await asyncio.sleep(2)

    return {
        "type": "async",
        "thread": threading.current_thread().name,
    }

同期処理は、time.sleep(2)で現在のスレッドを2秒間停止します。FastAPIでは、通常の def で定義したAPIはスレッドプールで実行されるため、この待機によってイベントループ自体はブロックされません。

非同期処理は、await asyncio.sleep(2)で2秒間待機しますが、その間はイベントループに実行を譲り、ほかの処理を進められます。

また、今回は実行スレッドを確認するため、threading.current_thread().nameでスレッド名を取得しています。

app/main.py に次のimportを追加します。

from app.routers.execution_demo import router as execution_demo_router

続いて、既存の include_router() と同じ場所に追加してください。

app.include_router(execution_demo_router)

Dev Container内のプロジェクトルートでFastAPIを起動します。

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

Swagger UIを開いて次の2つのAPIを実行してください。

• GET /execution-demo/sync
• GET /execution-demo/async

想定される結果は次の通りです。
※ スレッド名は実行環境によって異なる場合があります。

同期API：

{
    "type": "sync",
    "thread": "AnyIO worker thread"
}

非同期API：

{
    "type": "async",
    "thread": "MainThread"
}

どちらも応答まで約2秒かかります。

今回の結果から分かるのは、次の違いです。

• def：FastAPIがスレッドプールを利用し、ワーカースレッドでAPIを実行する。
• async def：イベントループが動作しているメインスレッドで実行し、await の間にほかの処理へ実行を譲ることができる。

つまり、単に async def を使ったからといって、1つのAPIの処理速度が速くなるわけではありません。

CORS（Cross-Origin Resource Sharing）は、異なるオリジンのWebページからAPIなどのリソースへアクセスする際に、ブラウザがその応答へのアクセスを許可するか判断するための仕組みです。

例えば、NuxtなどのフロントエンドからFastAPIを呼び出す場面で関係してきます。

オリジンとは、以下の3つを組み合わせたものです。

• プロトコル（http / https）
• ホスト（localhost / example.comなど）
• ポート番号（3000 / 8000など）

例えば、以下の構成を考えます。

＜フロントエンド＞
Nuxt
http://localhost:3000 
           ↓↑ APIリクエスト
＜バックエンド＞
FastAPI
http://localhost:8000

どちらも localhost ですが、ポート番号が異なります。

そのため、ブラウザでは別のオリジンとして扱われます。

このように、3つの要素が一致している必要があります。（公式ドキュメント）

ブラウザには、同一オリジンポリシー（Same-Origin Policy）というセキュリティ上の制約があります。

例えば、悪意のあるWebサイトを開いた際に、そのサイトのJavaScriptが別のWebサイトの情報を自由に読み取れてしまうと危険です。

そのため、ブラウザは原則として、異なるオリジンへのJavaScriptからのアクセスを制限します。

ただし、実務ではフロントエンドとAPIを別のドメインやポートで動かすことがよくあります。

そこで、API側が「このオリジンからのアクセスは許可する」とブラウザに伝える仕組みがCORSです

フロントエンドからFastAPIのAPIを呼び出す場合を考えます。

なお、リクエストの種類によっては、ブラウザが実際のAPIリクエストより前に OPTIONS メソッドで許可を確認します。

これをプリフライトリクエスト（Preflight Request）と呼びます。例えば、JSONを送信する一般的なPOSTリクエストでは発生します。

すべてのリクエストでプリフライトが発生するわけではありません。

CORSは、API自体へのアクセスを完全に禁止する認証・認可の仕組みではありません。

例えば、ブラウザのJavaScriptではCORSエラーが発生しても、通常のcurlやサーバー側のPythonプログラムからはAPIを呼び出せる場合があります。

つまり、CORSの役割は主に ブラウザが別オリジンからの応答をJavaScriptに公開してよいか制御することです。

ユーザー認証やアクセス権限のチェックは別途実装する必要があります。

FastAPIでは CORSMiddleware を利用します。

このMiddlewareで、許可するオリジン、HTTPメソッド、リクエストヘッダーなどを指定できます。

実際の設定は次の項目で行います。

今回は、FastAPIにCORSの設定を追加し、許可したオリジンからのアクセスと、許可していないオリジンからのアクセスで、レスポンスヘッダーがどのように変わるか確認します。

今回は以下の構成を想定します。

フロントエンドからFastAPIへアクセスできるように、CORSMiddleware を設定します。

app/main.pyの既存のコードを残したまま、以下の部分を追加してください。

from fastapi.middleware.cors import CORSMiddleware

続いて、app = FastAPI() の直後に追加します。

# CORSでアクセスを許可するオリジン
origins = [
    "http://localhost:3000",
]

# CORSの設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

既存の include_router() はそのまま残してください。

コードのポイント

今回は http://localhost:3000 だけを許可しています。

allow_headers=["*"] は、すべてのリクエストヘッダーを許可する設定です。

なお、今回はCookieなどの認証情報を伴うクロスオリジン通信は扱わないため、allow_credentials はデフォルトの False のままにしています。

まずは、CORSの設定が正しく反映されているかを確認しましょう。

FastAPIを起動した状態で、Dev Container内の別のターミナルから以下を実行してください。

curl -i \
  -H "Origin: http://localhost:3000" \
  http://localhost:8000/execution-demo/async

前の項目で作成した非同期APIを使用しています。

今回のポイントは以下です。

-H "Origin: http://localhost:3000"

リクエストヘッダーに Origin を指定することで、ブラウザが送るCORSリクエストの一部を再現しています。

レスポンスヘッダーに、以下が含まれていればOKです。

HTTP/1.1 200 OK
access-control-allow-origin: http://localhost:3000

これにより、FastAPIが指定したオリジンを許可するレスポンスヘッダーを返していることが確認できます。

※ curl 自体はCORS制限を適用しません。今回は、ブラウザが判断に使用するヘッダーを確認しています。

今度は、許可していない http://localhost:4000 を指定してみましょう。

Dev Container内のターミナルで実行してください。

curl -i \
  -H "Origin: http://localhost:4000" \
  http://localhost:8000/execution-demo/async

以下が返ってくればOKです。

HTTP/1.1 200 OK

ただし、先ほど表示されていた以下のヘッダーは返されません。

access-control-allow-origin: http://localhost:4000

ここがポイントです。

通常のGETリクエストでは、許可されていないオリジンでもAPI自体は処理され、200が返る場合があります。

しかし、許可を示すCORSヘッダーがないため、ブラウザはその応答をJavaScriptに公開しません。

最後に、プリフライトリクエストを確認しましょう。

ブラウザが実際のAPIリクエストを送信する前に、OPTIONS メソッドを使って、HTTPメソッドやリクエストヘッダーが許可されているか確認する仕組みです。

今回は curl でそのリクエストを再現します。

Dev Container内で実行してください。

curl -i -X OPTIONS \
  -H "Origin: http://localhost:3000" \
  -H "Access-Control-Request-Method: GET" \
  -H "Access-Control-Request-Headers: X-Demo" \
  http://localhost:8000/execution-demo/async

今回のリクエストは、「http://localhost:3000 から、X-Demo というヘッダー付きのGETリクエストを送信してもよいか？」という問い合わせです。

レスポンスに以下のような内容が含まれればOKです。

HTTP/1.1 200 OK
access-control-allow-origin: http://localhost:3000
access-control-allow-methods: GET, POST, PUT, DELETE
access-control-allow-headers: X-Demo

これらのヘッダーが返されれば、今回要求したオリジン・メソッド・ヘッダーが許可されていることを確認できます。

なお、これは事前確認なので、実際のGET APIの処理は実行されません。

今回は、これまでに学んだ「環境変数による設定管理」と「CORS」を組み合わせた課題です。

現在、main.py には次のようにオリジンを直接記述しています。

origins = [
    "http://localhost:3000",
]

この設定を変更して、許可するオリジンを .env から取得できるようにしてください。

＜実装要件＞

1. .env に FRONTEND_ORIGIN を追加し、値を http://localhost:4000 にする。
2. config.py の Settings に設定項目を追加する。
3. main.py のCORS設定で settings から取得した値を利用する。
4. 既存のAPIやDB接続の動作に影響を与えない。

今回は複数オリジンの管理までは不要です。1つのオリジンを設定できればOKです。

実装後、curl で次の結果になることを確認してください。

あわせて、localhost:4000 からのプリフライトリクエストが正常に許可されることも確認してください。

＜実装レビュー＞

＜動作確認レビュー＞

DB接続も問題なく動作しているため、STEP 7の自力課題は完了です！