【第3回】AWS CDK × Python インフラ自動化入門 — 環境分離・スタック分割・カスタムコンストラクトで本番構成に近づける

前回（第2回）はコンストラクトの考え方を整理して、S3やDynamoDBといったリソースを1つのスタックにガシガシ書いて動かすところまでやりました。動いたときはうれしかったんですが、書いてる途中から「このファイル、そのうち1000行超えるな……」という不安がずっとありました。

そこで今回は、スタックを分けて、dev / prod を切り替えられるようにして、API Gateway + Lambda のセットをカスタムコンストラクトに切り出す、という回です。テーマ的には「AWS CDK（Python）での API Gateway 構築」がメインになります。

• 環境ごとの設定をどこに置くか（cdk.json の context か、Python の dataclass か）
• スタックをどの粒度で分けるか、分けたあとのリソース参照
• API Gateway + Lambda を自作コンストラクトにまとめる書き方
• REST API のステージ設定（アクセスログ・スロットリング）まで含めた実装例

今回のディレクトリ構成

最終的にこうなりました。前回はスタック1個だったので急に増えて見えますが、中身は薄いです。

my_app/
├── app.py
├── cdk.json
├── config.py              # 環境ごとの設定
├── my_constructs/
│   └── api_function.py    # Lambda + ロググループをまとめた自作コンストラクト
├── stacks/
│   ├── data_stack.py      # DynamoDB
│   └── api_stack.py       # API Gateway + Lambda
└── lambda/
    └── handler.py

ディレクトリ名を constructs にしたら CDK 本体の constructs パッケージと衝突して ImportError で数分悩みました。ここ、環境や import の状況によっては踏みやすい罠な気がします。

分割の基準は「ライフサイクルが違うものは別スタック」でいいらしいです。DBは消えたら困るけどAPIは何度でも作り直したい、という感覚は自分にも分かるので、まずそこで線を引きました。逆に Lambda と API Gateway は一緒に更新されることがほぼ100%なので同じスタックです。スタックを増やしすぎるとデプロイ順や参照関係が面倒になるので、最初は2〜3個で十分かなと。

環境ごとの設定を1か所に寄せる

ここが一番迷いました。AWS CDK で環境分離をやる方法、調べると少なくとも3パターン出てきます。

• cdk.json の context に環境ごとの値を書いて --context env=prod で切り替える
• Python の dataclass で設定オブジェクトを作って app.py で渡す
• 環境ごとに AWS アカウントを分けて、コードは同じままアカウントだけ変える

3つ目が本来の王道らしいんですが、個人開発でアカウントを2つ運用するのは正直しんどい。ということで dataclass ベースにして、環境名だけ context から受け取る形にしました。IDEの補完が効くのと、typo で None が飛んでくる事故が減るのが決め手です。

# config.py
from dataclasses import dataclass
from aws_cdk import aws_logs as logs

@dataclass(frozen=True)
class EnvConfig:
    env_name: str
    account: str
    region: str
    log_retention: logs.RetentionDays
    throttle_rate: int
    protect_data: bool

ENVS = {
    "dev": EnvConfig("dev", "111122223333", "ap-northeast-1",
                     logs.RetentionDays.ONE_WEEK, 10, False),
    "prod": EnvConfig("prod", "111122223333", "ap-northeast-1",
                      logs.RetentionDays.THREE_MONTHS, 100, True),
}

最初は保持日数を int で持たせようとしたんですが、RetentionDays は enum なので任意の数値からは作れません。素直に enum をそのまま設定値として持たせるのが楽でした。設定ファイルに CDK の型が混ざるのは気持ち悪い気もするんですが、変換用の dict を書くより短いので妥協。

app.py はこうなります。

# app.py
import aws_cdk as cdk
from config import ENVS
from stacks.data_stack import DataStack
from stacks.api_stack import ApiStack

app = cdk.App()
env_name = app.node.try_get_context("env") or "dev"
cfg = ENVS[env_name]

aws_env = cdk.Environment(account=cfg.account, region=cfg.region)

data = DataStack(app, f"{env_name}-data", cfg=cfg, env=aws_env)
ApiStack(app, f"{env_name}-api", cfg=cfg, table=data.table, env=aws_env)

cdk.Tags.of(app).add("env", env_name)
cdk.Tags.of(app).add("managed-by", "cdk")

app.synth()

cdk.Tags.of(app) でアプリ配下の全リソースにタグを付けられるのは知らなかったので、地味に感動しました。あとから「これ何のリソースだっけ」を防げるのは大きいです。

デプロイは cdk deploy --all --context env=dev、本番は --context env=prod に変えるだけ。スタック名に環境名が入っているので同一アカウントでも衝突しません。ただ、cdk.json の context に env を書き込んでしまうと引数を忘れたことに気づけないので、自分はデフォルト dev のまま cdk.json には書かない運用にしています。ここは好みが分かれそう。

Lambda まわりをカスタムコンストラクトにまとめる

Lambda を1個作るたびに「関数定義 → ロググループ → 環境変数 → タイムアウト」と同じことを書くのが面倒なので、まとめます。カスタムコンストラクトは Construct を継承するだけなので、想像よりずっと簡単でした。

# my_constructs/api_function.py
from aws_cdk import Duration, RemovalPolicy, aws_lambda as _lambda, aws_logs as logs
from constructs import Construct

class ApiFunction(Construct):
    def __init__(self, scope, id, *, entry, handler, retention,
                 environment=None, memory=256, timeout=10):
        super().__init__(scope, id)

        self.log_group = logs.LogGroup(
            self, "Logs",
            retention=retention,
            removal_policy=RemovalPolicy.DESTROY,
        )

        self.function = _lambda.Function(
            self, "Fn",
            runtime=_lambda.Runtime.PYTHON_3_13,
            code=_lambda.Code.from_asset(entry),
            handler=handler,
            memory_size=memory,
            timeout=Duration.seconds(timeout),
            environment=environment or {},
        )

ポイントはログ保持まわり。Lambda の log_retention プロパティは、裏でカスタムリソース用の Lambda を作る仕組みで、テンプレートに余計なリソースが増えがちです。自分で LogGroup を定義しておけば cdk synth の出力もすっきりするし、RemovalPolicy も自分で握れます。

コンストラクトの中で self.function を公開しているのは、外から権限付与（grant_*）したいからです。本当は権限もコンストラクト側に閉じ込めたほうが行儀がいい気がするんですが、今の規模だと過剰かなと思ってこのままにしています。

REST API を組む：LambdaRestApi と proxy=False

API Gateway の REST API は aws_cdk.aws_apigateway にあります。一番手っ取り早いのが LambdaRestApi で、ハンドラを渡すだけで全リクエストを Lambda に丸投げする API ができます。ドキュメントにもある通り proxy=False を指定すると、リソースとメソッドを自分で定義する形に切り替わります。個人的にはこっちが好みでした。ルーティングが CDK のコードだけ見て分かるので。

# stacks/api_stack.py
from aws_cdk import Stack, CfnOutput, aws_apigateway as apigw, aws_logs as logs
from my_constructs.api_function import ApiFunction

class ApiStack(Stack):
    def __init__(self, scope, id, *, cfg, table, **kwargs):
        super().__init__(scope, id, **kwargs)

        fn = ApiFunction(
            self, "ItemsFn",
            entry="lambda",
            handler="handler.main",
            retention=cfg.log_retention,
            environment={"TABLE_NAME": table.table_name, "ENV": cfg.env_name},
        )
        table.grant_read_write_data(fn.function)

        access_logs = logs.LogGroup(self, "ApiAccessLogs", retention=cfg.log_retention)

        api = apigw.LambdaRestApi(
            self, "Api",
            handler=fn.function,
            proxy=False,
            rest_api_name=f"{cfg.env_name}-items-api",
            deploy_options=apigw.StageOptions(
                stage_name=cfg.env_name,
                throttling_rate_limit=cfg.throttle_rate,
                throttling_burst_limit=cfg.throttle_rate * 2,
                logging_level=apigw.MethodLoggingLevel.INFO,
                data_trace_enabled=False,
                metrics_enabled=True,
                access_log_destination=apigw.LogGroupLogDestination(access_logs),
                access_log_format=apigw.AccessLogFormat.json_with_standard_fields(),
            ),
            default_cors_preflight_options=apigw.CorsOptions(
                allow_origins=apigw.Cors.ALL_ORIGINS,
                allow_methods=["GET", "POST", "OPTIONS"],
            ),
        )

        items = api.root.add_resource("items")
        items.add_method("GET")
        items.add_method("POST")
        items.add_resource("{item_id}").add_method("GET")

        CfnOutput(self, "ApiUrl", value=api.url)

add_method("GET") だけで引数なしなのは、LambdaRestApi がデフォルト統合として渡した Lambda を持っているからです。メソッドごとに別の Lambda に振り分けたい場合は RestApi を使って apigw.LambdaIntegration(other_fn) を第2引数に渡す形になります。今回は1関数で受けて中でルーティングする「モノリシックLambda」構成にしました。関数を分けるほうがコールドスタート的には有利らしいんですが、まだそこまで気にする規模じゃないので。

ステージ名の扱いで少しハマった

stage_name に環境名を入れると URL が https://xxxx.execute-api.../dev/items になります。dev と prod で API 自体が別なのにパスにも環境名が入るのは冗長かもしれない、と後から思いました。ステージを環境の代わりに使う設計（1つの API に dev / prod ステージを並べる）もあるみたいですが、それだとスタック分離の意味が薄れるので、自分はステージ名を v1 固定にする方向に変えるつもりです。この辺はまだ結論が出ていません。

スタックをまたいだ参照で注意すること

table=data.table のようにオブジェクトをそのまま渡すと、CDK が勝手に CloudFormation の Output と Fn::ImportValue を作ってくれます。これは便利なんですが、いったん export されると 参照元を消すまで export を削除できない ので、あとで構成を変えたときに「削除できないスタック」が生まれることがあります。

回避策として、参照される側のスタックで先に値を固定しておく方法があります。

# DataStack 側
self.export_value(self.table.table_arn)

正直この挙動はまだ完全に理解できていません。個人開発だと「困ったら消して作り直す」で済んでしまうので、痛い目を見るまで実感が湧かないタイプの知識だと思っています。とはいえ本番相当のスタックを気軽に消せないのは分かるので、頭の片隅には置いておきたい。

そういえば、スタックを分けてから cdk diff がすごく読みやすくなりました。以前は全リソースの差分が一気に流れてきて目が滑っていたのが、「api スタックだけ変わってる」と一目で分かるのは精神衛生的に良いです。

デプロイして叩いてみる

cdk synth --context env=dev
cdk deploy --all --context env=dev

curl https://xxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/items

--all を付けると依存関係を見て data → api の順に流してくれます。逆にスタックを1つだけ更新したいときは cdk deploy dev-api のようにスタック名を指定。ワイルドカードも使えるので環境によっては cdk deploy "dev-*" みたいな指定でも動くようです（シェル側の解釈も絡むので注意）。

ちなみに HTTP API（apigatewayv2）で作る選択肢もあって、そっちのほうが安くて速いと言われています。ただ REST API 側にしかない機能（リクエストバリデーション、使用量プラン、WAF連携など）もあるので、今回は REST でそろえました。バージョンによってモジュールの分かれ方が変わっている印象があるので、v2 を使うときは公式ドキュメントで import 先を確認したほうが安全です。

今回わかったこと

スタック分割と環境分離は、やる前は「大規模プロジェクトの話でしょ」と思っていたんですが、実際にやってみると個人開発でも普通に効きました。特に効いたのは diff が読めるようになったことと、dev で壊しても prod に影響しない安心感です。カスタムコンストラクトについては、正直まだ「関数にまとめたのと何が違うんだ」感が残っていて、L3 コンストラクトのありがたさを実感するのはもう少し複雑な構成になってからかなと思っています。