前回（第1回）は、CDKの環境構築から cdk init → cdk bootstrap → 最初の cdk deploy までを一通り触って、「PythonでCloudFormationのテンプレートが吐き出される」という感覚をつかむところまでを紹介しました。今回はその続きで、実際にアプリで使うリソース（S3・DynamoDB・IAM）をどう書くか、という話です。
正直、第1回のスタックはほぼ空っぽだったので、ここからがCDKの本番かなと思っています。特にDynamoDBは自分がよく使う割にコンソールでポチポチしていた部分なので、コード化してみたら発見が多かったです。
この記事でわかること
- DynamoDBのテーブルを
TableV2で定義する書き方（GSI・TTL・削除ポリシーまで） - S3バケットを「安全側のデフォルト」で作るときの指定
- IAMを手書きポリシーではなく
grant_*メソッドで組む考え方 - リソースをConstructで部品化する実装パターン
今回作る構成
お題は毎回同じだと飽きるので、「フォームから投稿されたデータをDynamoDBに入れて、添付ファイルはS3に置く」くらいのミニ構成にしました。Lambdaは一旦あと回しにして、今回はストレージ層とIAMだけです。
- DynamoDB テーブル 1本（シングルテーブル設計っぽく pk/sk 構成）
- S3 バケット 1つ（添付ファイル置き場）
- 上記2つにアクセスするための IAM ロール
ディレクトリは第1回のまま、my_app/my_app_stack.py に書き足していきます。
DynamoDBのテーブル定義はTableV2で書く
CDKでDynamoDBを書くとき、ググると dynamodb.Table を使う記事がまだ多いんですが、今は dynamodb.TableV2 のほうが推奨されているようです。中身としては CloudFormation の AWS::DynamoDB::GlobalTable にマッピングされていて、単一リージョンでもそっちで作られます。
from aws_cdk import Stack, RemovalPolicy, Duration
from aws_cdk import aws_dynamodb as dynamodb
from constructs import Construct
class MyAppStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs):
super().__init__(scope, construct_id, **kwargs)
table = dynamodb.TableV2(
self, "AppTable",
partition_key=dynamodb.Attribute(
name="pk", type=dynamodb.AttributeType.STRING
),
sort_key=dynamodb.Attribute(
name="sk", type=dynamodb.AttributeType.STRING
),
billing=dynamodb.Billing.on_demand(),
time_to_live_attribute="expires_at",
removal_policy=RemovalPolicy.DESTROY, # ここ注意
)
ポイントは、table_name を書いていないところです。指定しないとCDKが自動で名前を付けてくれます。最初は「名前くらい自分で決めたい」と思って明示していたんですが、テーブル名を固定すると同じアカウント内でdev/prodを分けたときに衝突するので、今は基本的にCDK任せにしています。アプリ側には環境変数でテーブル名を渡せば済む話でした。
削除ポリシーは最初に決めておく
上のコードで RemovalPolicy.DESTROY に「ここ注意」とコメントを付けたのは、TableV2のデフォルトが RETAIN（スタックを消してもテーブルは残る）だからです。検証用なら消えてほしいので DESTROY にしていますが、これを本番用のスタックに書いたまま cdk destroy したら普通にデータが消えます。自分は検証環境で一度やらかして、まあ検証だったのでダメージはゼロでしたが、心臓には悪かったです。
データを守る系だと、ポイントインタイムリカバリも入れておくと安心です。TableV2では point_in_time_recovery_specification のほうを使います。
point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
point_in_time_recovery_enabled=True,
),
GSIはリストではなくメソッドで足せる
GSIの追加はコンストラクタ引数でもできますが、add_global_secondary_index() で後付けするほうが読みやすい気がしています。
table.add_global_secondary_index(
index_name="gsi1",
partition_key=dynamodb.Attribute(
name="gsi1pk", type=dynamodb.AttributeType.STRING
),
sort_key=dynamodb.Attribute(
name="gsi1sk", type=dynamodb.AttributeType.STRING
),
projection_type=dynamodb.ProjectionType.INCLUDE,
non_key_attributes=["status", "updated_at"],
)
ProjectionType.ALL にすると楽ですがストレージが増えるので、必要な属性だけ INCLUDE するのが定石らしいです。ちなみに INCLUDE を指定したのに non_key_attributes を書き忘れると synth の時点で怒られることが多いようです。これはCDKが守ってくれてる感じがして地味に好きなポイントです。
あと、GSIの追加自体は既存テーブルに対して無停止でできますが、削除・作り直しになる変更（パーティションキーの変更など）はテーブルごとリプレイスされます。cdk diff で「replace」と出たら一旦手を止めるのが安全です。
S3は「明示的に安全側」を書いておく
S3側はこんな感じにしました。
from aws_cdk import aws_s3 as s3
bucket = s3.Bucket(
self, "AttachmentBucket",
block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
encryption=s3.BucketEncryption.S3_MANAGED,
enforce_ssl=True,
versioned=True,
lifecycle_rules=[
s3.LifecycleRule(
noncurrent_version_expiration=Duration.days(30),
abort_incomplete_multipart_upload_after=Duration.days(7),
)
],
removal_policy=RemovalPolicy.DESTROY,
auto_delete_objects=True,
)
最近のS3は新規作成バケットで「ブロックパブリックアクセス」が自動で有効化される方向になっているので、block_public_access は書かなくても実質そうなることが多いです。それでも書いているのは、コードを読んだ人が「これはprivateなバケットなんだな」と一目でわかるほうがいいかなと思ったからです。CDKは省略できる引数が多いぶん、意図を残すために「あえて書く」判断がけっこう大事な気がしています。
auto_delete_objects=True は中身ごと消してくれる便利オプションです（スタック削除前にバケット内オブジェクトを消すための仕組みが入ります）。スタックを synth すると急に見覚えのないリソースが増えていて「なんだこれ」となったので一応共有しておきます。
IAMはgrantメソッドに任せる
個人的に、CDKで一番「これは楽だ」と思ったのがここです。ポリシーをJSONで手書きせずに、リソース側のメソッドを呼ぶだけで済みます。
from aws_cdk import aws_iam as iam
role = iam.Role(
self, "AppRole",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name(
"service-role/AWSLambdaBasicExecutionRole"
)
],
)
table.grant_read_write_data(role)
bucket.grant_put(role)
bucket.grant_read(role)
これだけで、テーブルへの最小権限寄りのポリシーが生成されます（ケースによってはインデックス（.../index/*）向けの権限も含まれることがあります）。ただ、このへんはCDKのバージョンや構成でも差が出やすいので、最終的には生成されたIAMポリシー（CloudFormation出力やコンソール）を確認するのが安全です。自分で書いていた頃、GSIに対する権限を書き忘れて AccessDeniedException と数十分にらめっこしたことがあるので、この自動化はかなりありがたいです。
細かい条件（dynamodb:LeadingKeys による行レベル制限とか）はgrantメソッドでは表現できないので、そういったケースは iam.PolicyStatement を素直に書きます。全部grantでやろうとして無理やり継ぎ足すより、そこだけ手書きするほうが読みやすいと思っています。
role.add_to_policy(
iam.PolicyStatement(
actions=["dynamodb:Query"],
resources=[table.table_arn],
conditions={
"ForAllValues:StringEquals": {
"dynamodb:LeadingKeys": ["USER#${aws:userid}"]
}
},
)
)
そろそろStackが太ってきたのでConstructに切る
ここまでで1つのスタッククラスが100行近くなってきました。CDKでは自前のConstructを作って、リソースのまとまりを部品化できます。
class StorageConstruct(Construct):
def __init__(self, scope: Construct, construct_id: str, *, retain: bool = True):
super().__init__(scope, construct_id)
policy = RemovalPolicy.RETAIN if retain else RemovalPolicy.DESTROY
self.table = dynamodb.TableV2(...)
self.bucket = s3.Bucket(...)
def grant_all(self, grantee: iam.IGrantable) -> None:
self.table.grant_read_write_data(grantee)
self.bucket.grant_read_write(grantee)
スタック側は storage = StorageConstruct(self, "Storage", retain=False) と書くだけになるので、だいぶスッキリします。ただ、Constructに切り出すと論理IDのパスが変わるのでリソースが作り直される点は要注意です（cdk diff で気づけます）。「最初からConstructで書いておけばよかった」と後から思うやつですね。
スタックを分けるかConstructで分けるか問題については、正直まだ自分の中で答えが出ていません。ライフサイクルが違うもの（DBとアプリ）はスタックを分けたほうがいい、という意見をよく見るんですが、分けると分けたでクロススタック参照が増えて削除しづらくなるらしく、その辺は今後自分でもちゃんと踏んで学びたいです。
デプロイ前に必ずdiffを見る
ここまで書いたら、cdk diff です。
cdk diff
cdk deploy --require-approval any-change
diffの出力で [-] や「may be replaced」と出ているところだけ先に読む、という運用に落ち着きました。CDKは書くのが簡単すぎて、うっかりリソースを消す変更もサラッと書けてしまうので、ここは面倒でも毎回やっています。CIに cdk diff だけ流すGitHub Actionsを組んでおくと、PRの時点で差分が見えて便利でした。
まとめ
AWS CDK × Pythonでストレージとセキュリティ周りの基本パターンをまとめてみました。
- DynamoDBは
TableV2で最新の書き方を。削除ポリシーは環境に応じて慎重に選ぶ - S3は「安全側のデフォルト」を明示的に書いておくと、後で読む人にも優しい
grant_*メソッドでポリシーを自動生成できるのがCDKの大きなメリット。複雑な条件は手書きポリシーとハイブリッドで- Constructで部品化すると、スタックが巨大化するのを防げる
cdk diffは毎回見るクセをつけておくと、本番トラブルを減らせます
📚 シリーズ「AWS CDK × Python インフラ自動化入門」（第2回 / 全4回）
← 前回の記事: 前回の記事はこちら
→ 次回の記事: 公開後にリンクが追加されます
参考になったらクリックしてもらえると嬉しいです!
