個人的なFastAPI バックエンド構成 決定版

はじめに

FastAPIを使ったバックエンド開発でよく使っているお気に入りの構成があるので紹介したいと思います。

特徴

• 中規模開発向け構成
• FactoryMethodパターン
• Poetryで依存関係管理
• PydanticでDTO実装

ディレクトリ構成

.
├── core/
│   ├── base/
│   │   ├── abstractions/
│   │   │   └── __init__.py ... sharedの抽象クラスをここでまとめる
│   │   ├── providers/ ... プロバイダー抽象クラス
│   │   └── utils/
│   │       └── __init__.py ... sharedの純粋関数をまとめる
│   ├── main/
│   │   ├── abstractions.py
│   │   ├── api/
│   │   │   └── **_router.py ... FastAPIでルーティングを定義
│   │   ├── app.py ... FastAPIを管理
│   │   ├── app_entry.py ... メインファイル(uvicornで呼び出す)
│   │   ├── assembly/
│   │   │   ├── builder.py ... ここで全機能の初期化を行う
│   │   │   └── factory.py ... クラスはここでインスタンス化する
│   │   └── services/ ... サービスクラス
│   │       └── **_service.py
│   └── providers/　... プロバイダークラス
│       └── **Provider.py
├── makefile ... コマンド用
├── poetry.lock
├── pyproject.toml
├── shared/ ... 共通で使えるもの
    ├── abstractions/ ... 共通の抽象クラス
│   └── utils/ ... 純粋関数

実行の流れ

app_entry.py

uvicorn core.main.app_entry:appとして指定するファイルです。
ここではBuilderをインスタンス化してbuild()を実行します。

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from core.main.assembly.builder import Builder

@asynccontextmanager
async def lifespan(app: FastAPI):
    builder = Builder()
    _app = await builder.build()

    app.router.routes = _app.app.routes
    app.middleware = _app.app.middleware

    yield
    # 終了後に実行したい処理

app = FastAPI(lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

builder.py

Builderでは処理全体の初期化を行います。
Factoryの実行や、Serviceのインスタンス化を行い、Routerクラスに渡します。

from typing import Any, Type
from core.main.abstractions import Services
from core.main.app import App
from core.main.assembly.factory import ProviderFactory
from core.main.services.githubapp_service import AuthService
from core.main.api.auth_router import AuthRouter

class Builder:
    _SERVICES = ["auth"]

    def __init__(self):
        pass

    async def _create_providers(self, provider_factory: Type[ProviderFactory]) -> Any:
        factory = provider_factory()
        return factory.create_providers()

    async def _create_services(self, service_params) -> Services:
        services = Builder._SERVICES
        service_instances = {}

        for service_type in services:
            service_class = globals()[f"{service_type.capitalize()}Service"]
            service_instances[service_type] = service_class(**service_params)

        return Services(**service_instances)

    async def build(self) -> App:
        provider_factory = ProviderFactory
        try:
            providers = await self._create_providers(
                provider_factory=provider_factory
            )
        except Exception as e:
            raise Exception(f"build error: {e}")

        service_params = {
            "providers": providers,
        }

        services = await self._create_services(service_params)

        routers = {
            "auth_router": AuthRouter(
                providers=providers,
                services=services
            ).get_router()
        }

        return App(**routers)

app.py

AppクラスではFastAPIの初期化を行い、app_entry.pyに返します。

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from core.main.api.auth_router import AuthRouter

class App:
    def __init__(
        self,
        auth_router: AuthRouter
    ):
        self.auth_router = auth_router
        self.app = FastAPI()
        self._setup_routes()

    def _setup_routes(self):
        # ルータークラスを追加する
        self.app.include_router(self.auth_router)
        # self.app.include_router(self.**_router)

        # OpenAPI仕様書
        @self.app.get("/openapi_spec", include_in_schema=False)
        async def openapi_spec():
            return get_openapi(
                title="OpenAPI仕様書",
                version="1.0.0",
                routes=self.app.routes,
            )

    async def serve(self, host: str = "0.0.0.0", port: int = 7272):
        import uvicorn

        config = uvicorn.Config(
            self.app,
            host=host,
            port=port,
        )
        server = uvicorn.Server(config)
        await server.serve()

factory.py

各Providerはここでインスタンス化されます。
サンプルコードなので無駄が多いように見えますが、実際にはここでProviderごとにconfig情報を渡したり、条件によって使用するProviderを切り替えたりします。

from core.providers.auth import AuthProvider
from core.main.abstractions import Provider

class ProviderFactory:
    def __init__(self) -> None:
        pass

    # ここでプロバイダーを作成する
    @staticmethod
    async def create_auth_provider(*args, **kwargs):
        return AuthProvider(*args, **kwargs)
    
    # @staticmethod
    # async def create_**_provider(*args, **kwargs):
    #     return **Provider(*args, **kwargs)

    async def create_provider(self):
        auth_provider = await self.create_auth_provider()
        # **_provider = await self.create_**_provider()

        return Providers(
            auth=auth_provider,
        )

auth_router.py

以下はAuthのRouterクラスの例です。

from fastapi import APIRouter, Body
from pydantic import EmailStr

class AuthRouter:
    def __init__(
        self,
        providers: Providers,
        services: Services,
    ):
        self.providers = providers
        self.services = services

        self.router = APIRouter()

        self._setup_routes()

    def get_router(self):
        return self.router

    def _setup_routes(self):
        @self.router.post("/register")
        async def register(
            # Pydanticのバリデーションに引っ掛かるとクライアントにエラーが返されます
            email: EmailStr = Body(..., description="メールアドレス"),
            password: str = Body(..., description="パスワード"),
            name: str = Body(..., description="名前"),
        ):
            # サービスを呼び出す
            await self.services.auth.register(email, password, name)

        @self.router.post("/login")
        async def login(
            email: EmailStr = Body(..., description="メールアドレス"),
            password: str = Body(..., description="パスワード"),
        ):
            pass

        @self.router.post("/logout")
        async def logout(request):
            pass

makefile

実行コマンドは長いのでmakefileで簡単にします。

run:
    poetry run uvicorn core.main.app_entry:app --reload

.vscode/settings.jsonのおすすめ設定

Poetryのvirtualenvのパスをvscodeに指定するところが肝です。
これを指定しないとpoetry addでインストールしたモジュールがvscode側で認識されません。

{
  "[python]": {
    "editor.defaultFormatter": "ms-python.autopep8",
    "editor.formatOnSave": true,
    "editor.codeActionsOnSave": {
      "source.organizeImports": "always"
    }
  },
  "isort.args": ["--profile", "black"],
  "makefile.configureOnOpen": false,
  "python.analysis.autoImportCompletions": true, // 勝手にインポートしてくれる
  "python.defaultInterpreterPath": "<Poetryで作られる仮想環境のdirを指定>",
  "python.analysis.typeCheckingMode": "basic" // 静的型チェック設定
}

.vscode/extensions.json

{
  "recommendations": [
    "ms-python.autopep8",
    "ms-python.vscode-pylance",
    "ms-python.isort",
  ]
}