FastAPIの環境構築

FastAPIの環境構築

FastAPIをインストールするには、以下の手順を実行します。まず、Pythonがインストールされていることを確認してください。Pythonがインストールされていない場合は、公式サイトからインストールしてください。

インストール

FastAPIをインストールするには、以下の手順を実行します。まず、Pythonがインストールされていることを確認してください。Pythonがインストールされていない場合は、公式サイトからインストールしてください。

仮想環境の作成

プロジェクトごとに仮想環境を作成することをお勧めします。以下のコマンドで仮想環境を作成します。

python3 -m venv myenv
source myenv/bin/activate

仮想環境を使用することで、プロジェクトごとに依存関係を管理しやすくなります。

FastAPIとUvicornのインストール

仮想環境がアクティブになった状態で、以下のコマンドを実行してFastAPIとUvicornをインストールします。

pip install fastapi uvicorn

このコマンドを実行することで、FastAPIとUvicornがインストールされます。インストールが完了したら、以下のコマンドでFastAPIのバージョンを確認できます。

pip show fastapi

依存関係の管理

プロジェクトの依存関係を管理するために、requirements.txtファイルを作成し、以下のように記述します。

fastapi
uvicorn

このファイルを使用して、他の開発者が同じ環境を再現できるようにします。

pip install -r requirements.txt

これにより、プロジェクトのセットアップが簡単になります。

基本構文

FastAPIの基本的な構文を紹介します。

エンドポイントの定義

FastAPIでは、エンドポイントを簡単に定義できます。以下に例を示します。

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    if item_id == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id, "q": q}

リクエストとレスポンス

FastAPIは、リクエストとレスポンスの処理を簡単に行えます。以下に例を示します。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
async def create_item(item: Item):
    return item

エラーハンドリング

FastAPIでは、エラーハンドリングも簡単に行えます。以下に例を示します。

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

APIモード専用

FastAPIを使用してAPIを開発する際の基本的な手順を紹介します。

APIの基本設定

FastAPIを使用する際には、以下のように基本的な設定を行います。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

サーバーの起動

Uvicornを使用してサーバーを起動します。以下のコマンドを実行します。

uvicorn main:app --reload

--reloadオプションを指定することで、コードの変更を自動的に検知してサーバーを再起動します。

APIドキュメントの確認

FastAPIは自動的にAPIドキュメントを生成します。サーバーが起動したら、http://localhost:8000/docsでSwagger UIを使用してAPIドキュメントを確認できます。

API開発のベストプラクティス

FastAPIを使用してAPIを開発する際のベストプラクティスを以下に示します。

ドキュメントの自動生成

FastAPIは、APIドキュメントを自動的に生成します。開発中にhttp://localhost:8000/docsでSwagger UIを使用してAPIドキュメントを確認し、エンドポイントの仕様を常に最新に保ちます。

バリデーションの活用

pydanticを使用して、リクエストデータのバリデーションを行います。これにより、データの整合性を保ち、エラーを未然に防ぐことができます。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

非同期処理の活用

FastAPIは非同期処理をサポートしています。async/awaitを使用して、非同期に処理を行い、パフォーマンスを向上させます。

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

テストの自動化

FastAPIアプリケーションのテスト自動化方法を以下に示します。

pytestを使用したテスト

pytestを使用して、FastAPIアプリケーションのテストを自動化します。

インストール

まず、pytestをインストールします。

pip install pytest

テストの実装

以下に、FastAPIアプリケーションのテストの例を示します。

from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

• TestClient: FastAPIアプリケーションをテストするためのクライアントです。
• assert: テストの期待値を確認します。

テストの実行

以下のコマンドを実行して、テストを実行します。

pytest

認証と認可の詳細

FastAPIでの認証と認可の詳細な実装方法を以下に示します。

OAuth2による認証

OAuth2を使用して、トークンベースの認証を実装します。

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

JWTによる認証

JWT（JSON Web Token）を使用して、トークンベースの認証を実装します。

from jose import JWTError, jwt

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # ユーザー認証ロジック
    access_token = jwt.encode({"sub": form_data.username}, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": access_token, "token_type": "bearer"}

ユーザーロールと権限管理

ユーザーロールや権限を管理して、アクセス制御を行います。

def get_current_user(token: str = Depends(oauth2_scheme)):
    # トークンのデコードとユーザーの取得
    return user

@app.get("/admin")
async def read_admin_data(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Not enough permissions")
    return {"admin_data": "..."}

セキュリティと認証

FastAPIでのセキュリティ対策や認証の実装方法を以下に示します。

OAuth2による認証

FastAPIは、OAuth2を使用した認証をサポートしています。以下に、OAuth2を使用した認証の例を示します。

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

JWTによる認証

JWT（JSON Web Token）を使用して、トークンベースの認証を実装します。以下に、JWTを使用した認証の例を示します。

from jose import JWTError, jwt

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # ユーザー認証ロジック
    access_token = jwt.encode({"sub": form_data.username}, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": access_token, "token_type": "bearer"}

セキュリティヘッダーの設定

CORS（Cross-Origin Resource Sharing）を設定して、セキュリティを強化します。

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

• allow_origins: 許可するオリジンを指定します。

CSRF対策

CSRF（Cross-Site Request Forgery）対策を実装します。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.middleware("http")
async def csrf_protect(request: Request, call_next):
    if request.method == "POST":
        token = request.headers.get("X-CSRF-Token")
        if not token or token != "expected_token":
            return JSONResponse(status_code=403, content={"detail": "CSRF token missing or incorrect"})
    response = await call_next(request)
    return response

セキュリティヘッダーの設定

セキュリティヘッダーを設定して、セキュリティを強化します。

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    return response

デプロイメント

FastAPIアプリケーションのデプロイメント方法を以下に示します。

Dockerを使用したデプロイメント

Dockerを使用して、FastAPIアプリケーションをコンテナ化し、デプロイします。

Dockerfileの作成

以下の内容でDockerfileを作成します。

FROM python:3.9

WORKDIR /app

COPY . /app

RUN pip install --no-cache-dir -r requirements.txt

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

Dockerイメージのビルド

以下のコマンドを実行して、Dockerイメージをビルドします。

docker build -t myfastapiapp .

Dockerコンテナの起動

以下のコマンドを実行して、Dockerコンテナを起動します。

docker run -d --name myfastapiapp -p 80:80 myfastapiapp

クラウドサービスへのデプロイ

FastAPIアプリケーションをAWS、GCP、Azureなどのクラウドサービスにデプロイすることも可能です。各クラウドサービスのドキュメントを参照して、適切な手順でデプロイを行います。

高度なルーティング

FastAPIでの高度なルーティングの設定方法を以下に示します。

パスパラメータ

パスパラメータを使用して、動的なルートを定義します。

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

• {user_id}: パスパラメータとして定義され、エンドポイントに渡されます。

クエリパラメータ

クエリパラメータを使用して、リクエストに追加の情報を渡します。

@app.get("/items/")
async def read_items(q: str = None):
    return {"q": q}

• q: str = None: クエリパラメータとして定義され、オプションでリクエストに含めることができます。

パスオペレーションの設定

パスオペレーションに追加の設定を行い、詳細なルーティングを実現します。

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    return {"item_id": item_id}

• response_model=Item: レスポンスモデルを指定し、返されるデータの型を定義します。

ミドルウェアの活用

FastAPIでのミドルウェアの設定方法を以下に示します。

ミドルウェアの追加

ミドルウェアを使用して、リクエストの前処理や後処理を行います。

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class SimpleMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # リクエストの前処理
        response = await call_next(request)
        # レスポンスの後処理
        return response

app.add_middleware(SimpleMiddleware)

• BaseHTTPMiddleware: ミドルウェアを作成するための基底クラスです。
• dispatch: リクエストとレスポンスの処理を行うメソッドです。

ログ記録の実装

リクエストとレスポンスのログを記録するミドルウェアを実装します。

import logging

logger = logging.getLogger("uvicorn")

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        logger.info(f"Request: {request.method} {request.url}")
        response = await call_next(request)
        logger.info(f"Response: {response.status_code}")
        return response

app.add_middleware(LoggingMiddleware)

バックグラウンドタスク

FastAPIでのバックグラウンドタスクの実装方法を以下に示します。

バックグラウンドタスクの追加

バックグラウンドタスクを使用して、リクエストの処理後に非同期でタスクを実行します。

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message + "\n")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification sent"}

• BackgroundTasks: バックグラウンドタスクを管理するためのクラスです。
• add_task: バックグラウンドで実行するタスクを追加します。

非同期タスクの管理

バックグラウンドタスクを使用して、非同期にタスクを管理します。これにより、リクエストの応答を待たずにタスクを実行できます。

WebSocketの使用

FastAPIでのWebSocketの使用方法を以下に示します。

WebSocketの基本設定

WebSocketを使用して、リアルタイム通信を実現します。

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

• WebSocket: WebSocket接続を管理するためのクラスです。
• accept: クライアントからの接続を受け入れます。
• receive_text: クライアントからのメッセージを受信します。
• send_text: クライアントにメッセージを送信します。

リアルタイム通信の実装

WebSocketを使用して、リアルタイムでデータを送受信します。これにより、チャットアプリケーションや通知システムなどを実現できます。

リアルタイムデータ処理

FastAPIでのリアルタイムデータ処理の実装方法を以下に示します。

WebSocketを使用したリアルタイムデータのストリーミング

WebSocketを使用して、リアルタイムでデータをストリーミングします。

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

• WebSocket: WebSocket接続を管理するためのクラスです。
• accept: クライアントからの接続を受け入れます。
• receive_text: クライアントからのメッセージを受信します。
• send_text: クライアントにメッセージを送信します。

リアルタイムデータ処理の実装

WebSocketを使用して、リアルタイムでデータを送受信します。これにより、チャットアプリケーションや通知システムなどを実現できます。

データベースとの連携

FastAPIでのデータベースとの連携方法を以下に示します。

SQLAlchemyを使用したデータベース操作

SQLAlchemyを使用して、データベースと連携します。

インストール

まず、SQLAlchemyをインストールします。

pip install sqlalchemy

データベースモデルの定義

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

データベース操作

from sqlalchemy.orm import Session

# ユーザーの作成
async def create_user(db: Session, user: User):
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

Tortoise ORMを使用したデータベース操作

Tortoise ORMを使用して、非同期でデータベースと連携します。

インストール

pip install tortoise-orm

データベースモデルの定義

from tortoise import fields, Tortoise, run_async
from tortoise.models import Model

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    email = fields.CharField(max_length=50, unique=True)

async def init():
    await Tortoise.init(
        db_url='sqlite://db.sqlite3',
        modules={'models': ['__main__']}
    )
    await Tortoise.generate_schemas()

run_async(init())

パフォーマンスの最適化

FastAPIアプリケーションのパフォーマンス最適化方法を以下に示します。

キャッシュの活用

キャッシュを使用して、データの取得を高速化します。fastapi-cacheを使用して、キャッシュを実装します。

インストール

pip install fastapi-cache

キャッシュの設定

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.inmemory import InMemoryBackend

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    FastAPICache.init(InMemoryBackend())

@app.get("/items/{item_id}")
@FastAPICache(expire=60)
async def read_item(item_id: int):
    return {"item_id": item_id}

• FastAPICache: キャッシュを管理するためのクラスです。
• InMemoryBackend: メモリ内キャッシュを使用します。

非同期処理の最適化

非同期処理を最適化して、パフォーマンスを向上させます。async/awaitを使用して、非同期に処理を行います。

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

非同期タスクのスケジューリング

FastAPIでの非同期タスクのスケジューリング方法を以下に示します。

APSchedulerを使用したタスクスケジューリング

APSchedulerを使用して、定期的に非同期タスクを実行します。

インストール

pip install apscheduler

タスクの定義とスケジューリング

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger

app = FastAPI()
scheduler = AsyncIOScheduler()

async def scheduled_task():
    print("Task executed")

@app.on_event("startup")
async def startup_event():
    scheduler.add_job(scheduled_task, IntervalTrigger(seconds=10))
    scheduler.start()

@app.on_event("shutdown")
async def shutdown_event():
    scheduler.shutdown()

• AsyncIOScheduler: 非同期タスクをスケジュールするためのスケジューラです。
• IntervalTrigger: タスクを定期的に実行するためのトリガーです。

国際化とローカライゼーション

FastAPIでの国際化とローカライゼーションの実装方法を以下に示します。

多言語対応の設定

fastapi-i18nを使用して、多言語対応を実現します。

インストール

pip install fastapi-i18n

設定ファイルの作成

localesディレクトリに、各言語の翻訳ファイルを作成します。

locales/
  en.json
  ja.json

翻訳ファイルの例

en.json:

{
  "greeting": "Hello"
}

ja.json:

{
  "greeting": "こんにちは"
}

国際化の設定

from fastapi import FastAPI
from fastapi_i18n import I18nMiddleware

app = FastAPI()

app.add_middleware(I18nMiddleware, default_locale="en", locales_dir="locales")

@app.get("/greet")
async def greet():
    return {"message": app.i18n.t("greeting")}

• I18nMiddleware: 国際化を管理するためのミドルウェアです。
• t: 翻訳キーを使用して、翻訳されたメッセージを取得します。

APIバージョニング

FastAPIでのAPIバージョニングの実装方法を以下に示します。

バージョン管理のベストプラクティス

APIのバージョン管理を行うことで、後方互換性を保ちながら新機能を追加できます。

パスベースのバージョニング

URLパスにバージョン番号を含める方法です。

from fastapi import FastAPI

app = FastAPI()

@app.get("/v1/items/{item_id}")
async def read_item_v1(item_id: int):
    return {"item_id": item_id, "version": "v1"}

@app.get("/v2/items/{item_id}")
async def read_item_v2(item_id: int):
    return {"item_id": item_id, "version": "v2"}

ヘッダーベースのバージョニング

HTTPヘッダーにバージョン情報を含める方法です。

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int, accept_version: str = Header(None)):
    if accept_version == "v1":
        return {"item_id": item_id, "version": "v1"}
    elif accept_version == "v2":
        return {"item_id": item_id, "version": "v2"}
    else:
        raise HTTPException(status_code=400, detail="Invalid API version")

エラーハンドリングの拡張

FastAPIでのエラーハンドリングの拡張方法を以下に示します。

カスタムエラーハンドラー

カスタムエラーハンドラーを実装して、特定のエラーに対するカスタムレスポンスを返します。

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": f"Oops! {exc.detail}"},
    )

ログの記録

エラー発生時にログを記録して、デバッグを容易にします。

import logging

logger = logging.getLogger("uvicorn.error")

@app.exception_handler(Exception)
async def generic_exception_handler(request, exc):
    logger.error(f"Unexpected error: {exc}")
    return JSONResponse(
        status_code=500,
        content={"message": "Internal Server Error"},
    )

ドキュメントのカスタマイズ

FastAPIの自動生成ドキュメントのカスタマイズ方法を以下に示します。

OpenAPI仕様の拡張

OpenAPI仕様を拡張して、APIドキュメントをカスタマイズします。

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI()

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="Custom title",
        version="2.5.0",
        description="This is a very custom OpenAPI schema",
        routes=app.routes,
    )
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Swagger UIのカスタマイズ

Swagger UIをカスタマイズして、APIドキュメントの外観を変更します。

from fastapi.openapi.docs import get_swagger_ui_html

@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
    return get_swagger_ui_html(
        openapi_url=app.openapi_url,
        title=app.title + " - Custom Docs",
        swagger_favicon_url="https://example.com/favicon.ico"
    )

• get_swagger_ui_html: Swagger UIのHTMLを生成するための関数です。

ファイルアップロードとダウンロード

FastAPIでのファイルアップロードとダウンロードの実装方法を以下に示します。

ファイルアップロード

ファイルをアップロードするためのエンドポイントを定義します。

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

• UploadFile: アップロードされたファイルを表すクラスです。
• File: ファイルをアップロードするための依存関係を定義します。

ファイルダウンロード

ファイルをダウンロードするためのエンドポイントを定義します。

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

@app.get("/downloadfile/{file_path}")
async def download_file(file_path: str):
    return FileResponse(path=file_path, filename="downloaded_file.txt")

• FileResponse: ファイルをレスポンスとして返すためのクラスです。

APIゲートウェイの設定

FastAPIをAPIゲートウェイとして使用する方法を以下に示します。

複数のマイクロサービスを統合

FastAPIを使用して、複数のマイクロサービスを統合し、APIゲートウェイとして機能させます。

ルーティングの設定

各マイクロサービスへのルーティングを設定します。

from fastapi import FastAPI, Request
import httpx

app = FastAPI()

@app.api_route("/service1/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy_service1(request: Request, path: str):
    url = f"http://service1/{path}"
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method=request.method,
            url=url,
            headers=request.headers,
            content=await request.body()
        )
    return response

• httpx.AsyncClient: 非同期HTTPクライアントを使用して、他のサービスにリクエストを転送します。

APIゲートウェイの利点

• セキュリティの強化: APIゲートウェイを通じて、セキュリティポリシーを一元管理します。
• 負荷分散: リクエストを複数のサービスに分散させ、負荷を軽減します。
• 監視とロギング: APIゲートウェイを通じて、リクエストの監視とロギングを行います。

デプロイメントのベストプラクティス

FastAPIアプリケーションのデプロイメントにおけるベストプラクティスを以下に示します。

コンテナ化

Dockerを使用して、FastAPIアプリケーションをコンテナ化し、移植性を向上させます。

Dockerfileの作成

以下の内容でDockerfileを作成します。

FROM python:3.9

WORKDIR /app

COPY . /app

RUN pip install --no-cache-dir -r requirements.txt

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

CI/CDパイプラインの構築

GitHub Actionsを使用して、CI/CDパイプラインを構築し、自動デプロイを実現します。

GitHub Actionsの設定

以下の内容で.github/workflows/deploy.ymlを作成します。

name: Deploy

on:
  push:
    branches:
      - main

jobs:
  build:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
    - name: Build and push Docker image
      run: |
        docker build -t myfastapiapp .
        docker push myfastapiapp

• Docker: アプリケーションをコンテナ化して、どこでも動作するようにします。
• GitHub Actions: CI/CDパイプラインを構築して、自動デプロイを実現します。

モニタリングとロギング

FastAPIアプリケーションのモニタリングとロギングの設定方法を以下に示します。

PrometheusとGrafanaを使用したモニタリング

PrometheusとGrafanaを使用して、アプリケーションのパフォーマンスをモニタリングします。

Prometheusの設定

Prometheusを使用して、メトリクスを収集します。

prometheus.yml

scrape_configs:
  - job_name: 'fastapi'
    static_configs:
      - targets: ['localhost:8000']

Grafanaの設定

Grafanaを使用して、メトリクスを可視化します。Prometheusをデータソースとして追加し、ダッシュボードを作成します。

ロギングの設定

Pythonのloggingモジュールを使用して、アプリケーションのログを記録します。

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("uvicorn")

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    logger.info(f"Fetching item with id: {item_id}")
    return {"item_id": item_id}

• logging: アプリケーションの動作を記録し、デバッグや監視に役立てます。

スケーラビリティの向上

FastAPIアプリケーションのスケーラビリティを向上させるための方法を以下に示します。

クラウドサービスを利用したスケーリング

AWS、GCP、Azureなどのクラウドサービスを利用して、アプリケーションをスケーリングします。

AWS Elastic Beanstalk

AWS Elastic Beanstalkを使用して、アプリケーションを自動スケーリングします。

# Elastic Beanstalk CLIを使用してデプロイ
eb init -p python-3.9 my-fastapi-app
eb create my-fastapi-env

GCP App Engine

GCP App Engineを使用して、アプリケーションをスケーリングします。

# app.yamlを作成してデプロイ
runtime: python39

# デプロイ
gcloud app deploy

コンテナオーケストレーション

Kubernetesを使用して、コンテナ化されたアプリケーションをオーケストレーションします。

Kubernetesの設定

Kubernetesを使用して、アプリケーションをスケーリングします。

deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-deployment
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fastapi
  template:
    metadata:
      labels:
        app: fastapi
    spec:
      containers:
      - name: fastapi
        image: myfastapiapp
        ports:
        - containerPort: 80

• replicas: アプリケーションのインスタンス数を指定して、スケーリングを行います。

カスタムミドルウェアの作成

FastAPIでのカスタムミドルウェアの作成方法を以下に示します。

カスタムミドルウェアの実装

リクエストの前処理や後処理を行うカスタムミドルウェアを作成します。

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # リクエストの前処理
        response = await call_next(request)
        # レスポンスの後処理
        return response

app.add_middleware(CustomMiddleware)

• BaseHTTPMiddleware: ミドルウェアを作成するための基底クラスです。
• dispatch: リクエストとレスポンスの処理を行うメソッドです。

ミドルウェアの活用例

• ログ記録: リクエストとレスポンスのログを記録します。
• 認証: リクエストの認証を行います。
• キャッシュ: レスポンスをキャッシュして、パフォーマンスを向上させます。

セキュリティ強化の詳細

FastAPIでのセキュリティ強化方法をさらに詳しく解説します。

JWTによる認証

FastAPIは、JWT（JSON Web Token）を使用した認証をサポートしています。

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid credentials")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

APIバージョニングの詳細

FastAPIでのAPIバージョニング方法をさらに詳しく解説します。

バージョン付きエンドポイント

from fastapi import FastAPI

app = FastAPI()

@app.get("/api/v1/items/")
async def read_items_v1():
    return {"version": "v1"}

@app.get("/api/v2/items/")
async def read_items_v2():
    return {"version": "v2"}

エラーハンドリングの拡張の詳細

FastAPIでのエラーハンドリングの拡張方法をさらに詳しく解説します。

カスタムエラーハンドラーの詳細

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": f"Oops! {exc.detail}"},
    )

ドキュメントのカスタマイズの詳細

FastAPIでのドキュメントのカスタマイズ方法をさらに詳しく解説します。

OpenAPIのカスタマイズの詳細

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI()

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="Custom title",
        version="2.5.0",
        description="This is a very custom OpenAPI schema",
        routes=app.routes,
    )
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi