【全2回】AWS Lambda x FastAPIによるPythonモダンAPI開発のすゝめ 2

はじめに

ラクスルグループのノバセルで新卒2年目のエンジニアをしています田村(tamtam)です。

第1回では、AWS Lambda x FastAPIによるPythonモダンAPI開発を実現する上で役立つであろう1. 開発環境の構築で使用したツール、2. 開発に活用したPythonライブラリについて紹介していきました。

そして本記事ではある第2回では3. アーキテキチャ及びディレクトリ構造について紹介していきます！💪

この記事を通じて、同じような課題を抱える他の開発者の皆さんに役立つ情報を提供できればと思います。

それでは詳細について見ていきましょう！

対象読者

• Pythonを使用して新しいプロジェクトを開始しようとしている人

• Python開発に役立つリンターやフォーマッターなどの効率的なライブラリを探している人

• Pythonでの開発に際して参考にしたいアーキテクチャを探している人

• AWS LambdaのPython環境をサーバーレス実行プラットフォームとして検討している人

• FastAPIを用いた開発を検討している人

あまり説明しないこと

• FastAPIやLambda自体の深い説明

前提とするバージョン

• Python: 3.10 　(2023年6月15日時点でのAWS Lambdaの最新バージョン)

参考となるレポジトリ

今回の記事に際して、筆者が作成したレポジトリを共有します。

github.com

なお、ソースコードは逐次更新しているため、本書の記述内容と差異がある可能性をご理解の上読んでいただきますと幸いです。

3. アーキテキチャ及びディレクトリ構造

今回の開発ではDomain層がInfrastructure層に依存させないオニオンアーキテクチャを採用しました。

また、FastAPI作者のテンプレートをはじめとするリポジトリを参考にし、各層で具体的にどういったモノを書くのかを決めていきました。

オニオンアーキテクチャを採用

オニオンアーキテクチャとは

オニオンアーキテクチャはドメイン駆動設計(DDD)といった設計のなかで代表されるアーキテクチャの1つです。

レイヤードアーキテクチャから、依存関係逆転の原則を用いてドメイン層とインフラ層の依存関係を逆転させたのがオニオンアーキテクチャです。

引用: ドメイン駆動設計 モデリング/実装ガイド

オニオンアーキテクチャでは、特定の技術（RDBMSやNoSQLなど）に依存しない形でドメイン層を独立したレイヤーとして設計することで、技術的変更に柔軟に対応する能力を実現しています。

引用: ドメイン駆動設計のアーキテクチャ

誕生の背景

オニオンアーキテクチャが生まれる前の一般的なアーキテクチャは、Ruby on Railsなどで見られる「プレゼンテーション層」「ビジネスロジック層」「データアクセス層」を備えた3層アーキテクチャでした。

この3層アーキテクチャでは、ビジネスロジック層にユースケースとドメイン知識が集中し、責任過多（低凝集）となる問題に直面しました。

さらに、ドメイン知識をモデルクラスに定義してしまうと、その知識がデータアクセス層に浸透（密結合）する問題もありました。

これを改善するために、ビジネスロジック層をアプリケーション（ユースケース）層とドメイン層に分割したレイヤードアーキテクチャが生まれました。

しかし、このレイヤードアーキテクチャでは、ドメイン層がインフラ層に依存しまいます。そのため、レイヤー間の高い結合度の問題が続いていました。

特に、DDD（ドメイン駆動設計）では、ドメイン層を他の層から独立させることが推奨されます。これは特定の技術に依存せず、モデルを逐次改良するための要件です。

これらの課題を解決するために、ドメイン層をインフラ層から切り離す目的で、依存関係逆転の原則を適用したオニオンアーキテクチャが提案されました。

依存関係逆転の原則の活用

実践ドメイン駆動設計を引用すると依存関係逆転の原則とは以下のようになります。

上位レベルのモジュールは下位レベルのモジュールに依存してはならない。どちらのモジュールも抽象に依存すべきである

抽象は実装に依存してはならない。 実装が抽象に依存すべきである

『実践ドメイン駆動設計』P.119

オニオンアーキテクチャでは、リポジトリのインターフェースをドメイン層で定義し、その具体的な実装クラスをインフラ層で設定します。

これにより、ドメイン層がインフラ層に依存する関係性を解消し、インフラ層がドメイン層で定義されているインターフェースに依存する形となります。

依存関係の方向

採用理由

ノバセルではすでにいくつかのPythonプロジェクトでクリーンアーキテクチャを試みてきました。

その経験から、基本的に同じ思想を持つオニオンアーキテクチャの導入はそれほど難しくないと感じました。

クリーンアーキテクチャの開発では、都度インターフェースを設けることによるボイラープレート過多の問題がありました。しかし今回の開発では、リポジトリをインターフェースと実装に分ける位で、その問題が軽減されると考えました。

一方で、3層アーキテクチャでも十分なのではないかという意見もありました。

しかし、このAPIは一定の拡張性が必要とされています。またシンプルな構成にすると詳細が曖昧になり、その結果、何をどこに記述すべきかが不明確になると考えました。

そのため、ある程度の方針が示され、参考にしやすいオニオンアーキテクチャが見通しを良くすると判断しました。

オニオンアーキテクチャやその基盤となるDDDの書き方については、参考となる記事などがあるため、学習しながら進めることができました。

参考になった記事

オニオンアーキテクチャを採用する上で参考になった記事を共有します。 Pythonで実際にコーディングする上で読んだリソースは後述します。

• ドメイン駆動設計のアーキテクチャ

• オニオンアーキテクチャにておいて、ドメイン層とアプリケーション層の責務はどう違うのか[DDD] - little hands' lab

ディレクトリ構造

全体の構成

実際のディレクトリ構造とは微妙に異なるかもしれませんが、以下に示すような形を想定しています。参考にしていただければ幸いです。

├── api # presentation layer
│   ├── __init__.py
│   └── v1
│       ├── __init__.py
│       ├── endpoints
│       │   ├── __init__.py
│       │   └── <specific name>
│       └── router.py
├── compose.yml
├── container_config # DI container
│   ├── __init__.py
│   ├── di_container.py
│   └── module
│       ├── __init__.py
│       └── repository_module.py
├── core 
│   ├── __init__.py
│   ├── config.py
│   ├── logger
│   │   ├── __init__.py
│   │   ├── api_logger.py
│   │   └── logging_context_route.py
│   └── singleton.py
├── domain # domain layer
│   └── <specific name>
│       ├── service # business logic which is not written in entity and value object
│       ├── entity # mutable object
│       ├── i_repository # interface for repository(DIP)
│       └── value_object # immutable object
├── exceptions # custom exceptions
│   ├── core.py
│   ├── error_handle_middleware.py
│   └── error_messages.py
├── infrastructure # infrastructure layer
│    ├── repository_impl # implement repository interface
│    ├── model # ORM
│    ├──datasource 
├── main.py
├── schemas # DTO for using endpoint
├── scripts
│   └── compose_pytest
├── tests
├── usecase # application layer

aws-lambda-fastapi-onion-architecture-template

オニオンアーキテクチャとディレクトリの関係性を表すと、

オニオンアーキテクチャの各層	対応するディレクトリ
プレゼンテーション	api
ユースケース	usecase
ドメイン	domain
インフラ	infrastructure

となります。

そのほか、usecaseとapiとの受け渡しのデータ構造を提供するschemaやDIコンテナを実装するInjectorの制御を管理するcontainer_configなどがあります。

それぞれのディレクトリについて、データの処理の流れに沿って説明していきします。

api

apiディレクトリ配下では、大枠のルーティングの管理するrouterとそれをもとに呼ばれるパスオペレーション関数を提供します。

上記のような構成はFastAPI作者のテンプレートを参考にしています。

# router.py
v1_router = APIRouter()
v1_router.include_router(
    english_words.router, prefix="/english-words", tags=["english_words"]
)

# endpoints/english_words.py

router = APIRouter()

@router.get("/generate-meaning", response_model=MeaningGetResponse)
def get_meaning(word: str) -> MeaningGetResponse:
    usecase = DIContainer().resolve(GetMeaningUseCase)
    response = usecase.handle(meaning_get=MeaningGet(english_word=word)) 
    return response

schema

schemaの役割はapiとusecaseの間のデータ構造を提供する役割と、パスオペレーション関数のリクエストとレスポンスの構造を提供する役割を担います。

apiとusecaseの間のデータ構造を提供する役割

apiからusecaseへビジネスロジックに必要なパラメータ（例えば、タスク名、期日、ユーザーID）を送る際、1つのデータ構造にまとめるDTOを提供します。

その際、DTOではビジネスルールによらない基本的な制約（例えば、文字列の長さ、数値の範囲、必須フィールドの存在など）に基づいたバリデーションも行うようにします。

from pydantic import BaseModel, Field, HttpUrl


# Shared properties
class WordBase(BaseModel):
    english_word: str = Field(max_length=20)


class MeaningGet(WordBase):
    pass


class MeaningGetResponse(BaseModel):
    meaning: str

実際の使われ方は以下のようになります。

# endpoints/english_words.py

@router.get("/generate-meaning", response_model=MeaningGetResponse)
def get_meaning(word: str) -> MeaningGetResponse:
    usecase = DIContainer().resolve(GetMeaningUseCase)
    response = usecase.handle(meaning_get=MeaningGet(english_word=word))  # type: ignore
    return response

# usecase/english_word/get_meaning_usecase.py

class GetMeaningUseCase:
    @inject
    def __init__(self, meaning_repository: IMeaningOriginRepo) -> None:
        self.__meaning_repository = meaning_repository

    def handle(self, meaning_get: MeaningGet) -> MeaningGetResponse:
        english_word = EnglishWord(value=meaning_get.english_word)
        meaning = self.__meaning_repository.get(english_word)

        return MeaningGetResponse(meaning=meaning)

schemaはパスオペレーション関数のリクエストとレスポンスの構造を提供する役割

schemaによって定義されたDTOは、リクエストとレスポンスの構造を提供します。

FastAPIのパス操作関数は、リクエストに関してはBodyに対して、レスポンスについてはresponse_modelという引数に対して、Pydanticを継承したモデルを適用できます。

特に注目すべきは、response_modelについてです。これにより、指定されたモデルの構造に基づいてFastAPIがJSONにシリアライズして返すことが可能となります。

以下公式ドキュメントを引用

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()


@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
    return item

usecase

usecaseではドメイン層が公開するメソッドを組み合わせ、ユースケースを組み立てます。

from injector import inject

from domain.english_word.english_word import EnglishWord
from domain.english_word.i_meaning_repository import IMeaningOriginRepo
from schemas.english_word import MeaningGet, MeaningGetResponse


class GetMeaningUseCase:
    @inject
    def __init__(self, meaning_repository: IMeaningOriginRepo) -> None:
        self.__meaning_repository = meaning_repository

    def handle(self, meaning_get: MeaningGet) -> MeaningGetResponse:
        english_word = EnglishWord(value=meaning_get.english_word)
        meaning = self.__meaning_repository.get(english_word)

        return MeaningGetResponse(meaning=meaning)

domain

domainではコアとなるビジネスルールをもつエンティティやリポジトリのインターフェースを提供します。

# domain/entity/english_word.py

class EnglishWord(BaseModel, frozen=True):  # type: ignore
    value: str

    @validator("value")
    def is_english(cls, v: str) -> str:
        if not v.isascii():
            raise APIException(ErrorMessage.WORD_IS_NOT_ENGLISH)
        return v

# domain/english_word/i_meaning_repository.py

class IMeaningOriginRepo(metaclass=ABCMeta):
    @abstractmethod
    def get(self, english_word: EnglishWord) -> str:
        raise NotImplementedError()

infrastructure

infrastructureではドメイン層で定義したリポジトリの実装を提供します。

その際リポジトリの実装は、datasource内で定義したものを使用します。

# infrastructure/english_word/repository/meaning_dynamodb_repository_impl.py

from domain.english_word.english_word import EnglishWord
from domain.english_word.i_meaning_repository import IMeaningCacheRepo


class MeaningDynamodbRepositoryImpl(IMeaningCacheRepo):
    def get(self, english_word: EnglishWord) -> str:
        # Dynamodbから値が返る
        return "ギリシア文字の第11字母"

datasourceで書かれるモノは以下のようなコードを想定しています。

# Snowflakeをデータソースとして利用する場合の例


class Snowflake(Singleton):
    _conn = None

    @contextlib.contextmanager
    def cursor(self) -> Generator[SnowflakeCursor, None, None]:
        conn = self.__get_conn()
        cursor = conn.cursor()
        try:
            yield cursor
        except Exception as e:
            ApiLogger().error(error_obj=e)
            raise e
        finally:
            cursor.close()

    def __get_conn(self) -> SnowflakeConnection:
        if self._conn is not None:
            return self._conn
        
        self._conn = snowflake.connector.connect(
            user=SNOWFLAKE_USER,
            password=SNOWFLAKE_PASSWORD,
            account=SNOWFLAKE_ACCOUNT,
            warehouse=SNOWFLAKE_WH,
            database=SNOWFLAKE_DB,
            schema=SNOWFLAKE_SCHEMA,
        )
        return self._conn

core

coreではコンフィグやロギング、認証など共通で用いられるものを提供します。

ここでは例として、コンフィグの内容を記載します。

# core/config

import os

import yaml
from pydantic import BaseSettings, HttpUrl


# ref: https://qiita.com/ninomiyt/items/ee676d7f9b780b1d44e8
class Settings(BaseSettings):
    OPEN_AI_API_KEY: str
    SENTRY_DSN: HttpUrl | None
    ENV: str | None = None

    class Config:
        case_sensitive = True


# YAMLをロードする
# https://gist.github.com/ericvenarusso/dcaefd5495230a33ef2eb2bdca262011
def read_yaml(file_path: str) -> Settings:
    with open(file_path) as stream:
        config = yaml.safe_load(stream)

    return Settings(**config)


env = os.environ["ENV"]
settings = read_yaml(f"{os.getcwd()}/core/yaml_configs/{env}.yaml")

container_config

container_configではDIコンテナを提供します。

以下のようにインターフェースと実装のバインディングをします。

class RepositoryModule(Module):
    def configure(self, binder: Binder) -> None:
        binder.bind(IMeaningCacheRepo, to=impl.MeaningDynamodbRepositoryImpl)
        binder.bind(IMeaningOriginRepo, to=impl.MeaningOpenAIRepositoryImpl)

その後、DIコンテナを担当するクラスを提供させます。

class DIContainer(Singleton):
    MODULES = [RepositoryModule]
    injector: Injector = Injector(MODULES)

    # container_config.get()に引数を渡すと依存関係を解決してインスタンスを生成する
    def resolve(self, cls: object) -> object:
        return self.injector.get(cls)

呼ばれ方は以下のようになります。

# endpoints/english_words.py

@router.get("/generate-meaning", response_model=MeaningGetResponse)
def get_meaning(word: str) -> MeaningGetResponse:
    usecase = DIContainer().resolve(GetMeaningUseCase)
    response = usecase.handle(meaning_get=MeaningGet(english_word=word))  # type: ignore
    return response

exception

カスタム例外クラスやアプリケーションで起きた全体の例外処理を取得するエラー処理のミドルウェアを提供します。

基本的に以下のリポジトリのコードを参考に作成しています。興味ある方はご覧ください。 github.com

参考にしたもの

Pythonで実際にコーディングする上で参考にしたリソースを共有します。

• GitHub - tiangolo/full-stack-fastapi-postgresql: Full stack, modern web application generator. Using FastAPI, PostgreSQL as database, Docker, automatic HTTPS and more.

• GitHub - takashi-yoneya/fastapi-mybest-template: This is a practical template using FastAPI. It is dockerized including DB (MySQL) and uvicorn. Package management and task runner are implemented using poetry. FastAPIを使用した実用的なテンプレートです。 DB(MySQL)とuvicornを含めてdocker化しています。 poetryを使用して、パッケージ管理およびタスクランナーを実装しています。

• GitHub - teamhide/fastapi-layered-architecture: FastAPI Layered Architecture

• GitHub - iktakahiro/dddpy: Python DDD Example and Techniques

• FastAPI(Python)+DDD(ドメイン駆動開発)のポイント

• Python におけるドメイン駆動設計(戦術面)の勘どころ

• Python で学ぶ実践的なドメイン駆動設計とレイヤードアーキテクチャ / DDD and Onion Architecture in Python - Speaker Deck

• How to get started DDD & Onion-Architecture in Python web application

まとめ

第2回では、AWS Lambda x FastAPIによるPythonモダンAPI開発を実現する上で役立つであろう 3. アーキテキチャ及びディレクトリ構造について紹介していきました。

この記事を通じて、同じような課題を抱える他の開発者の皆さんに役立つ情報を提供できればと幸いです。

次回の記事としては、Snowflake ✖️ Pythonネタで何か書こうと思います！

他方、以前書いた「DWHからRDSへのデータロード処理の高速化」という記事も興味あれば見ていただけますと嬉しいです💪