Cloud Functions の横断的関心事をデコレーターに集約する

こんにちは！

広告事業本部でリードウェブアプリケーションエンジニアをしている森田です！

今回は Cloud Functions（第2世代）を複数本運用する過程で生まれた @functions_endpoint デコレーターの設計についてご紹介します。モノレポで54本もの Cloud Functions を管理するようになり、各関数に繰り返し書いていた定型処理をデコレーターに集約することで、コードの見通しが大きく改善しました。

モノレポで Cloud Functions を管理する工夫については前回の記事で紹介しているのでよければあわせてご覧ください。

blog.engineer.adways.net

本記事で使用しているバージョンは以下のとおりです。

Python 3.13
Pydantic 2.11
functions-framework 3.8
Flask 3.1

はじめに
Before / After
解決したかった問題
parse_request デコレーター：リクエストを Pydantic モデルに変換する
functions_endpoint デコレーター：ログ・例外処理を横断的に担う
全54関数への適用
設計上の工夫
まとめ

はじめに

Cloud Functions が増えるにつれて、ある問題が顕在化してきました。具体的には、リクエストのパース・バリデーション・ログ出力・例外処理といった定型処理を、各関数でコピーアンドペーストしてしまうことになる問題です。

最初は数本だったので気にならなかったのですが、数十本になると話が変わってきます。例えばログフォーマットを変えたいとなったとき、数十ファイルをすべて修正しなければなりません。（個別に関数での共通化はしていましたが、長いコードがボイラープレート化していました。）あるチームメンバーが修正を忘れれば、関数によってログの形式がバラバラになります。

この問題を解決するために作ったのが @functions_endpoint デコレーターです。

Before / After

まず、@functions_endpoint デコレーターを導入する前後でコードがどう変わるかを見てみます。

Before（デコレーターなし）

@functions_framework.http
def main(request: flask.Request) -> ResponseReturnValue:
    # メソッドチェック
    if request.method != "POST":
        return jsonify({"error": "Method Not Allowed"}), 405

    # リクエストパース
    data = request.get_json() or {}

    # 手動バリデーション
    user_id = data.get("user_id")
    name = data.get("name")
    if user_id is None or name is None:
        return jsonify({"error": "user_id and name are required"}), 400

    logger.info({"request": data})

    # ビジネスロジック
    try:
        result_id = do_something(user_id=user_id, name=name)
    except Exception as e:
        logger.exception("Unexpected Error")
        return str(e), 500

    logger.info({"response": {"id": result_id}})
    
    return flask.jsonify({"id": result_id}), 200

After（デコレーターあり）

class Request(BaseModel):
    user_id: int
    name: str


class Response(BaseModel):
    id: int


@functions_endpoint(Request, allowed_methods=["POST"])
def main(_, request_data: Request) -> ResponseReturnValue:
    result_id = do_something(
        user_id=request_data.user_id,
        name=request_data.name,
    )

    return flask.jsonify(Response(id=result_id).model_dump()), HTTPStatus.OK

パース・バリデーション・ログ出力・例外処理がデコレーターに移り、main 関数はビジネスロジックの呼び出しだけになりました。

解決したかった問題

Cloud Functions のエントリーポイントは Flask の request オブジェクトを受け取り、レスポンスを返す関数です。本来の責務は「リクエストを受け取る」「ビジネスロジックを呼び出す」「レスポンスを返す」の3つだけです。

しかし実際には、その周辺に定型処理が積み重なります。

JSON / form / query params が混在する場合のパース
Pydantic によるバリデーションとエラーレスポンスの返却
リクエスト・レスポンスのログ出力
未捕捉例外のスタックトレース記録と 500 レスポンス

これらを各関数に個別実装していると、54本に対して同じ変更が必要になったとき大変なことになります。また実装者によって微妙に書き方が違うと、コードの一貫性も崩れていきます。

`parse_request` デコレーター：リクエストを Pydantic モデルに変換する

デコレーターは2層に分けています。下層が parse_request で、HTTP リクエストを Pydantic モデルに変換することだけを担います。

# functions/utils/parse_request.py
def _is_method_error(request: flask.Request, allowed_methods: list[str] | None) -> bool:
    if allowed_methods is None:
        return False

    return request.method not in allowed_methods

def _request_to_dict(request: flask.Request) -> dict:
    """json, form, args をマージして辞書として返す"""
    data = {}
    if request.is_json:
        request_json = request.get_json()
        if request_json is not None:
            data.update(request_json)
    if request_form := request.form.to_dict():
        data.update(request_form)
    if request_args := request.args.to_dict():
        data.update(request_args)
    return data


def parse_request(pydantic_model: type[BaseModel], allowed_methods: list[str] | None = None) -> Callable:
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(request: flask.Request, *args, **kwargs) -> ResponseReturnValue:
            if _is_method_error(request, allowed_methods):
                return jsonify({"error": "Method Not Allowed"}), HTTPStatus.METHOD_NOT_ALLOWED
                
            try:
                data = _request_to_dict(request)
            except Exception as e:
                return jsonify({"error": "Bad Request", "details": str(e)}), HTTPStatus.BAD_REQUEST
                
            try:
                validated_data = pydantic_model.model_validate(data)
            except ValidationError as e:
                return jsonify({"error": "Validation error", "details": e.errors()}), HTTPStatus.BAD_REQUEST

            kwargs["request_data"] = validated_data
            
            return func(request, *args, **kwargs)

        return wrapper
    return decorator

JSON・form・query params を統合して1つの辞書にまとめてから model_validate に渡しています。システム全体としてこれらを分けて扱いたい要件が当面なさそうだったため、1つの辞書にマージすることで簡素化しています。バリデーション失敗は 400 を返して終了、成功すると request_data として kwargs に渡します。

HTTP メソッドのチェックもここで行っています。メソッドチェック・パース・バリデーションはいずれも「エントリーポイントに到達するリクエストの前処理」として一括して引き受けるということで、parse_request にまとめています。functions_endpoint と分離することで parse_request 単体のテストは独立して書くようにしています。

`functions_endpoint` デコレーター：ログ・例外処理を横断的に担う

上層が functions_endpoint です。 parse_request を内包しながら、ログ出力・例外処理・@functions_framework.http の適用をまとめて担います。

# functions/utils/functions_endpoint.py
def functions_endpoint(request_model: type[BaseModel], allowed_methods: list[str] | None = None) -> Callable:
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        @functions_framework.http
        @parse_request(request_model, allowed_methods)
        def wrapper(request: flask.Request, *args, **kwargs) -> ResponseReturnValue:
            workflows_execution_id = _parse_workflows_execution_id(request)
            if workflows_execution_id:
                set_workflows_execution_id(workflows_execution_id)

            logger = get_logger(__name__)
            request_data = kwargs.get("request_data")
            logger.info({"request": request_data.model_dump()})

            try:
                func_result = func(request, *args, **kwargs)
            except Exception as e:
                logger.exception("Unexpected Error")
                return str(e), HTTPStatus.INTERNAL_SERVER_ERROR

            logger.info({"response": func_result})
            return func_result

        return wrapper
    return decorator

X-Cloud-Workflow-Execution-ID ヘッダーの取り出しと ContextVar へのセットもここで行っています。これにより各関数の実装を意識せずに、全ログに Execution ID のラベルが自動付与されます。

全54関数への適用

結果として、すべての関数のエントリーポイントは次の形に統一されました。 After（デコレーターあり）の再掲です。

class Request(BaseModel):
    user_id: int
    name: str


class Response(BaseModel):
    id: int


@functions_endpoint(Request, allowed_methods=["POST"])
def main(_, request_data: Request) -> ResponseReturnValue:
    result_id = do_something(
        user_id=request_data.user_id,
        name=request_data.name,
    )

    return flask.jsonify(Response(id=result_id).model_dump()), HTTPStatus.OK

Request と Response の Pydantic モデルで入出力を宣言し、ビジネスロジックを呼ぶだけです。新しく横断的な処理を追加したくなったときは、デコレーターに1箇所追記するだけで全54関数に適用されます。

設計上の工夫

`@wraps(func)` を忘れずにつける

functools.wraps を使うことで元の関数名（__name__）や docstring が保持されます。つけ忘れるとデコレートされた関数はすべて wrapper という名前になり、ログなどで関数を区別できなくなります。ログやデバッガーで関数を特定する際に main として認識されるのは地味に重要です。

`parse_request` と `functions_endpoint` を分離している理由

2つに分けているのは、parse_request 単体でテストできるようにするためです。デコレーターが1つに統合されていると、パースの単体テストもエンドポイント全体の振る舞いを経由しないといけなくなります。

`@functions_framework.http` の位置

デコレーターは下から上に適用されるため、実行順は functions_framework.http → parse_request → wrapper 本体の順になります。 functions_framework が Flask の Request オブジェクトに変換してから parse_request が受け取れるよう、この順序にしています。

まとめ

横断的な関心事をデコレーターに集約することで、各 Cloud Functions のエントリーポイントを自身の責務だけに集中させることができました。 54本すべてが同じパターンで書かれているので、コードレビューもしやすく、新しいメンバーが関数を追加するときの迷いも減っています。モノレポで Cloud Functions を管理しているメリットの1つにもなっています。 AIコーディングエージェントのためのカスタム指示にも書きやすい規約になっていると感じます。

Cloud Functions を複数本運用していて定型処理の重複に悩んでいる方の参考になれば幸いです！