AWS CDK（ver.2）PythonでS3の署名付きURLを取得する構成を実装してみました

AWS CDKを使ってS3の署名付きURLを取得するAPIを実装してみました。今回はアップロードとダウンロードの2パターンの署名付きURLを作成するAPIを実装しています。
署名付きURLについては、他の方が詳しく書かれていますので、そちらを参照ください。
署名付き URLってなんだろ？

全コードはGithubを参照してください。
Githubはここ

前提条件

• AWS CDK v.2 がインストールされていること
• Python 3.x がインストールされていること
• AWSアカウントがあり、AWS CLIが設定されていること

※Cloud9を使うとこの辺りがPassできるため、Cloud9を使って今回の記事の内容は作成しています。

構築手順

1. CDKアプリの初期化

先ずはCDKアプリの初期化を行います。

$ mkdir presigned-test
$ cd presigned-test
$ cdk init --language python

2.必要なパッケージをインストール

続いて必要パッケージのインストールになります。
ここでは、CDKアプリを初期化した際に作成された.venvを有効化しています。

$ source .venv/bin/activate
$ pip install -r requirements.txt

3. スタックの記述

import os

from aws_cdk import (
    Stack,
    aws_apigateway as apigateway,
    aws_lambda as lambda_,
    aws_s3 as s3,
)
from constructs import Construct

class PresignedTestStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # Test用のS3Bucketを作成
        bucket = s3.Bucket(self, 'TestBucket')
        
        # /upload ファイルアップロード用にS3の署名付きURLを取得する関数
        upload_function = lambda_.Function(
            self,
            'UploadFunction',
            code=lambda_.Code.from_asset(
                os.path.join('lambda', 'upload_function')
            ),
            handler='lambda_function.lambda_handler',
            runtime=lambda_.Runtime.PYTHON_3_9,
            environment={
                'BUCKET_NAME': bucket.bucket_name,
            },
            function_name='upload-bucket',
        )
        
        bucket.grant_write(upload_function)
        
        # /download ファイルダウンロード用にS3の署名付きURLを取得する関数
        download_function = lambda_.Function(
            self,
            'DownloadFunction',
            code=lambda_.Code.from_asset(
                os.path.join('lambda', 'download_function')
            ),
            handler='lambda_function.lambda_handler',
            runtime=lambda_.Runtime.PYTHON_3_9,
            environment={
                'BUCKET_NAME': bucket.bucket_name,
            },
            function_name='download-bucket',
        )
        
        bucket.grant_read(download_function)
        
        # APIを作成
        rest_api = apigateway.RestApi(
            self,
            'RestApi',
            deploy=True,
            deploy_options=apigateway.StageOptions(stage_name='test'),
            rest_api_name='presigned-test-api',
        )
        
        # lambda integrationを作成
        upload_integration = apigateway.LambdaIntegration(
            handler=upload_function,
            integration_responses=[
                apigateway.IntegrationResponse(status_code='200')
            ],
        )
        download_integration = apigateway.LambdaIntegration(
            handler=download_function,
            integration_responses=[
                apigateway.IntegrationResponse(status_code='200')
            ],
        )
        
        # リソースを追加
        upload_resource = rest_api.root.add_resource(path_part='upload')
        download_resource = rest_api.root.add_resource(path_part='download')
        
        # Methodを追加
        upload_resource.add_method(
            http_method='GET',
            integration=upload_integration,
            method_responses=[
                apigateway.MethodResponse(status_code='200')
            ],
        )
        download_resource.add_method(
            http_method='GET',
            integration=download_integration,
            method_responses=[
                apigateway.MethodResponse(status_code='200')
            ],
        )

4. Lambda関数の記述

※今回は検証のため、ファイル名等をハードコーディングしていますのでご了承ください。

先ずはUploadするための署名付きURLを発行するためのLambda関数を記述します。

/lambda/upload/lambda_function.py

import boto3
import json
import os

def lambda_handler(event, context):
    bucket_name = os.environ['BUCKET_NAME']
    
    s3_client = boto3.client('s3')
    
    # 署名付きURLを取得
    put_url = s3_client.generate_presigned_url(
        ClientMethod='put_object',
        Params={'Bucket': bucket_name, 'Key': 'test.pptx'},
        ExpiresIn=600,
        HttpMethod='PUT'
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps(
            {
                'put_url': put_url,
            }
        )
    }

次にDownloadするための署名付きURLを発行するLambda関数を記述する。

/lambda/download/lambda_function.py

import boto3
import json
import os

def lambda_handler(event, context):
    bucket_name = os.environ['BUCKET_NAME']
    
    s3_client = boto3.client('s3')
    
    # 署名付きURLを取得
    get_url = s3_client.generate_presigned_url(
        ClientMethod='get_object',
        Params={'Bucket': bucket_name, 'Key': 'test.pptx'},
        ExpiresIn=600,
        HttpMethod='GET'
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps(
            {
                'get_url': get_url,
            }
        )
    }

5. CDKのデプロイ

cdk deploy

デプロイが完了すると、署名付きURLを作成するAPIにアクセスすることができます。

署名付きURLの検証

署名付きURLを使ったファイルアップロード検証

ファイルアップロード用の署名付きURLを取得

$ curl {APIGatewayのURL}/test/upload
{"put_url": "署名付きURL"}

署名付きURLを使ってファイルアップロード

$ curl -X PUT --upload-file {ファイル名} "署名付きURL"

アップロードが完了するとS3にファイルが存在することが確認できます。

署名付きURLを使ったファイルのダウンロード検証

ファイルダウンロード用の署名付きURLを取得

$ curl {APIGatewayのURL}/test/download
{"get_url": "署名付きURL"}

署名付きURLを使ってファイルダウンロード

$ curl -o {ファイル名} "署名付きURL"

ローカルにダウンロードされます

まとめ

この記事では、署名付きURLを使ってアップロードとダウンロードを実行するAPIを作成しました。API GatewayとLambdaを使ったサーバレスアプリの場合、6[MB]の制限に引っかかることがあるかと思いますので、その場合には候補の一つとしてご検討いただければよいのではないかと思います。
どなたかの参考になれば幸いです。