サーバーレスファイル処理アプリケーションを作成する
Lambda の最も一般的なユースケースの 1 つは、ファイル処理タスクを実行することです。例えば、Lambda 関数を使用して、HTML ファイルまたは画像から PDF ファイルを自動的に作成したり、ユーザーが画像をアップロードしたときにサムネイルを作成したりできます。
この例では、PDF ファイルが Amazon Simple Storage Service (Amazon S3) バケットにアップロードされると自動的に暗号化されるアプリケーションを作成します。このアプリを実装するには、以下のリソースを作成する必要があります。
-
ユーザーが PDF ファイルをアップロードするための S3 バケット
-
アップロードされたファイルを読み取り、暗号化されたパスワードで保護されたバージョンのファイルを作成する Python の Lambda 関数
-
Lambda が暗号化されたファイルに保存するための 2 番目の S3 バケット
また、AWS Identity and Access Management (IAM) ポリシーを作成して、S3 バケットで読み取りおよび書き込みオペレーションを実行するアクセス許可を Lambda 関数に付与します。
ヒント
Lambda を初めて使用する場合は、このサンプルアプリを作成する前にチュートリアル 最初の Lambda 関数を作成する から始めることをお勧めします。
AWS Management Console または AWS Command Line Interface (AWS CLI) を使用してリソースを作成して設定することで、アプリケーションを手動でデプロイできます。AWS Serverless Application Model (AWS SAM) を使用してアプリケーションをデプロイすることもできます。AWS SAM は、Infrastructure as Code (IaC) ツールです。IaC では、リソースを手動で作成するのではなく、コードに定義して自動的にデプロイします。
このサンプルアプリケーションをデプロイする前に、IaC で Lambda を使用する方法の詳細については、「Lambda と Infrastructure as code (IaC) の使用」を参照してください。
Lambda 関数のソースコードファイルを作成する
プロジェクトディレクトリに次のファイルを作成します。
-
lambda_function.py- ファイル暗号化を実行する Lambda 関数の Python 関数コード -
requirements.txt- Python 関数コードに必要な依存関係を定義するマニフェストファイル
以下のセクションを展開してコードを表示し、各ファイルの役割の詳細を確認してください。ローカルマシンにファイルを作成するには、以下のコードをコピーして貼り付けるか、aws-lambda-developer-guide GitHub レポジトリ
次のコードをコピーし、lambda_function.py という名前のファイルに貼り付けます。
from pypdf import PdfReader, PdfWriter import uuid import os from urllib.parse import unquote_plus import boto3 # Create the S3 client to download and upload objects from S3 s3_client = boto3.client('s3') def lambda_handler(event, context): # Iterate over the S3 event object and get the key for all uploaded files for record in event['Records']: bucket = record['s3']['bucket']['name'] key = unquote_plus(record['s3']['object']['key']) # Decode the S3 object key to remove any URL-encoded characters download_path = f'/tmp/{uuid.uuid4()}.pdf' # Create a path in the Lambda tmp directory to save the file to upload_path = f'/tmp/converted-{uuid.uuid4()}.pdf' # Create another path to save the encrypted file to # If the file is a PDF, encrypt it and upload it to the destination S3 bucket if key.lower().endswith('.pdf'): s3_client.download_file(bucket, key, download_path) encrypt_pdf(download_path, upload_path) encrypted_key = add_encrypted_suffix(key) s3_client.upload_file(upload_path, f'{bucket}-encrypted', encrypted_key) # Define the function to encrypt the PDF file with a password def encrypt_pdf(file_path, encrypted_file_path): reader = PdfReader(file_path) writer = PdfWriter() for page in reader.pages: writer.add_page(page) # Add a password to the new PDF writer.encrypt("my-secret-password") # Save the new PDF to a file with open(encrypted_file_path, "wb") as file: writer.write(file) # Define a function to add a suffix to the original filename after encryption def add_encrypted_suffix(original_key): filename, extension = original_key.rsplit('.', 1) return f'{filename}_encrypted.{extension}'
注記
このサンプルコードでは、暗号化されたファイル (my-secret-password) のパスワードが関数コードにハードコードされます。本番環境のアプリケーションでは、関数コードにパスワードなどの機密情報を含めないでください。AWS Secrets Manager を使用して、機密性の高いパラメータを安全に保存します。
Python 関数コードには 3 つの関数が含まれています。1 つは関数が呼び出されたときに Lambda が実行するハンドラー関数、他の 2 つは PDF 暗号化を実行するためにハンドラーが 呼び出す add_encrypted_suffix と encrypt_pdf という名前の別個の関数です。
関数が Amazon S3 によって呼び出されると、Lambda は呼び出しの原因となったイベントの詳細を含む JSON 形式のイベント引数を関数に渡します。この場合、情報には S3 バケットの名前と、アップロードされたファイルのオブジェクトキーが含まれます。Amazon S3 のイベントオブジェクトの形式の詳細については、「Lambda を使用した Amazon S3 イベント通知の処理」を参照してください。
次に、関数は AWS SDK for Python (Boto3) を使用して、イベントオブジェクトで指定された PDF ファイルをローカル一時ストレージディレクトリにダウンロードしてから、pypdf
最後に、この関数は Boto3 SDK を使用して暗号化されたファイルを S3 送信先バケットに保存します。
次のコードをコピーし、requirements.txt という名前のファイルに貼り付けます。
boto3 pypdf
この例では、関数コードには、標準の Python ライブラリに含まれていない依存関係が 2 つしかありません。SDK for Python (Boto3) と、関数が PDF 暗号化の実行に使用する pypdf パッケージです。
注記
SDK for Python (Boto3) のバージョンは Lambda ランタイムの一部として含まれているため、Boto3 を関数のデプロイパッケージに追加せずにコードが実行されます。ただし、関数の依存関係を完全に制御し、バージョン不一致による問題を回避するには、Python のベストプラクティスとして、関数のデプロイパッケージにすべての関数の依存関係を含めることをお勧めします。詳細については、「Python でのランタイム依存関係」を参照してください。
アプリをデプロイする
このサンプルアプリケーションのリソースは、手動または AWS SAM を使用して作成およびデプロイできます。実稼働環境では、AWS SAM のような IaC ツールを使用して、手動プロセスを使用せずにサーバーレスアプリケーション全体を迅速かつ繰り返しデプロイすることをお勧めします。
アプリを手動でデプロイするには:
-
ソースと送信先の Amazon S3 バケットを作成する
-
PDF ファイルを暗号化し、暗号化されたバージョンを S3 バケットに保存する Lambda 関数を作成する
-
オブジェクトがソースバケットにアップロードされたときに、関数を呼び出す Lambda トリガーを設定します。
開始する前に、ビルドマシンに Python
2 つの S3 バケットを作成する
まず、2 つの S3 バケットを作成します。1 つ目のバケットは、PDF ファイルをアップロードするソース バケットです。2 つ目のバケットは、関数を呼び出すときに暗号化されたファイルを保存するために Lambda が使用するバケットです。
実行ロールを作成する
実行ロールとは、AWS のサービス とリソースにアクセスする許可を Lambda 関数に付与する IAM ロールです。関数に Amazon S3 への読み取りおよび書き込みアクセスを許可するには、 AWS管理ポリシー AmazonS3FullAccess をアタッチします。
関数デプロイパッケージを作成する
関数を作成するには、関数コードとその依存関係を含むデプロイパッケージを作成します。このアプリケーションでは、関数コードは PDF 暗号化に別のライブラリを使用します。
デプロイパッケージを作成するには
-
以前に GitHub から作成またはダウンロードした
lambda_function.pyおよびrequirements.txtファイルを含むプロジェクトディレクトリに移動し、packageという名前の新しいディレクトリを作成します。 -
次のコマンドを実行して、
packageディレクトリのrequirements.txtファイルで指定された依存関係をインストールします。pip install -r requirements.txt --target ./package/ -
アプリケーション コードとその依存関係を含む .zip ファイルを作成します。Linux または MacOS では、コマンドラインインターフェイスから次のコマンドを実行します。
cd package zip -r ../lambda_function.zip . cd .. zip lambda_function.zip lambda_function.pyWindows では、任意の zip ツールを使用して、
lambda_function.zipファイルを作成します。lambda_function.pyファイルと依存関係が含まれるフォルダは、.zip ファイルのルートにインストールする必要があります。
また、Python 仮想環境を使用してデプロイパッケージを作成することもできます。「Python Lambda 関数で .zip ファイルアーカイブを使用する」を参照してください。
Lambda 関数を作成する
これで、前のステップで作成したデプロイパッケージを使用して Lambda 関数をデプロイします。
関数を呼び出すように Amazon S3 トリガーを設定する
ファイルをソース元のバケットにアップロードするときに Lambda 関数が実行されるようにするには、関数のトリガーを設定する必要があります。Amazon S3 トリガーは、コンソールまたは AWS CLI を使用して設定できます。
重要
この手順では、オブジェクトが S3 バケット内に作成されるたびに、関数を呼び出すようにバケットを設定します。この設定は、ソース元バケットのみで行うようにしてください。Lambda 関数が自身を呼び出した同じバケットにオブジェクトを作成する場合、関数が連続的にループして呼び出される
開始する前に、Docker
-
プロジェクトディレクトリで、次のコードをコピーして
template.yamlという名前のファイルに貼り付けます。プレースホルダーバケット名を置き換えます。-
ソースバケットの場合は、
amzn-s3-demo-bucketを S3 バケットの命名規則に準拠した任意の名前に置き換えます。 -
レプリケート先バケットの場合は、
amzn-s3-demo-bucket-encryptedを<source-bucket-name>-encryptedに置き換えます。ここで、<source-bucket>はレプリケート元バケットに選択した名前です。
AWSTemplateFormatVersion: '2010-09-09' Transform: AWS::Serverless-2016-10-31 Resources: EncryptPDFFunction: Type: AWS::Serverless::Function Properties: FunctionName: EncryptPDF Architectures: [x86_64] CodeUri: ./ Handler: lambda_function.lambda_handler Runtime: python3.12 Timeout: 15 MemorySize: 256 LoggingConfig: LogFormat: JSON Policies: - AmazonS3FullAccess Events: S3Event: Type: S3 Properties: Bucket: !Ref PDFSourceBucket Events: s3:ObjectCreated:* PDFSourceBucket: Type: AWS::S3::Bucket Properties: BucketName:amzn-s3-demo-bucketEncryptedPDFBucket: Type: AWS::S3::Bucket Properties: BucketName:amzn-s3-demo-bucket-encryptedAWS SAM テンプレートは、アプリケーション用に作成するリソースを定義します。この例では、テンプレートは
AWS::Serverless::Functionタイプを使用して Lambda 関数を定義し、AWS::S3::Bucketタイプを使用して 2 つの S3 バケットを定義します。テンプレートで指定されたバケット名はプレースホルダーです。AWS SAM を使用してアプリケーションをデプロイする前に、テンプレートを編集して、S3 バケットの命名規則を満たすグローバルに一意の名前でバケットの名前を変更する必要があります。このステップについては、「AWS SAM を使用してリソースをデプロイする」で詳しく説明します。Lambda 関数リソースの定義は、
S3Eventイベントプロパティを使用して関数のトリガーを設定します。このトリガーにより、ソースバケットにオブジェクトが作成されるたびに関数が呼び出されます。関数定義は、関数の実行ロールにアタッチする AWS Identity and Access Management (IAM) ポリシーも指定します。AWS 管理ポリシー
AmazonS3FullAccessは、Amazon S3 へのオブジェクトの読み取りと書き込みに必要なアクセス許可を関数に付与します。 -
-
lambda_function.py、requirements.txt、およびtemplate.yamlファイルを保存したディレクトリで、次のコマンドを実行します。sam build --use-containerこのコマンドによって、アプリケーションのビルドアーティファクトが収集され、それらをデプロイする場所を適切な形式で配置します。
--use-containerオプションを指定すると、Lambda のような Docker コンテナ内に関数がビルドされます。ここではこれを使用するため、ビルドを実行するためにローカルマシンに Python 3.12 をインストールする必要はありません。ビルドプロセス中に、AWS SAM はテンプレートの
CodeUriプロパティで指定した場所で Lambda 関数コードを検索します。この場合、現在のディレクトリをロケーション (./) として指定しました。requirements.txtファイルが存在する場合、AWS SAM はそれを使用して指定された依存関係を収集します。デフォルトでは、AWS SAM は関数コードと依存関係を含む .zip デプロイパッケージを作成します。PackageType プロパティを使用して、関数をコンテナイメージとしてデプロイすることもできます。 -
アプリケーションをデプロイし、AWS SAM テンプレートで指定された Lambda リソースと Amazon S3 リソースを作成するには、次のコマンドを実行します。
sam deploy --guided--guidedフラグを使用すると、AWS SAM にデプロイプロセスの手順が示されます。このデプロイでは、Enter キーを押してデフォルトのオプションをそのまま使用してください。
デプロイプロセス中に、AWS SAM でお使いの AWS アカウントに次のリソースが作成されます。
-
sam-appという名前の AWS CloudFormation スタック -
EncryptPDFという名前の Lambda 関数 -
template.yamlAWS SAM テンプレートファイルを編集したときに選択した名前の 2 つの S3 バケット -
名前形式
sam-app-EncryptPDFFunctionRole-の関数の IAM 実行ロール2qGaapHFWOQ8
AWS SAM がリソースの作成を完了すると、次のメッセージが表示されます。
Successfully created/updated stack - sam-app in us-east-2アプリケーションをテストする
アプリをテストするには、PDF ファイルをソースバケットにアップロードし、Lambda が送信先バケットに暗号化されたバージョンのファイルを作成することを確認します。この例では、コンソールまたは AWS CLI を使用して手動でテストするか、提供されたテストスクリプトを使用してテストできます。
本番環境のアプリケーションでは、ユニットテストなどの従来のテスト方法や手法を使用して、Lambda 関数コードが正しく機能していることを確認できます。ベストプラクティスは、実際のクラウドベースのリソースとの統合テストを実行する、提供されているテストスクリプトのようなテストを実行することです。クラウド環境における統合テストでは、インフラストラクチャが適切にデプロイされ、イベントが期待どおりに異なるサービス間で流れることを確認します。詳細については、「サーバーレス関数とアプリケーションをテストする方法」を参照してください。
Amazon S3 ソースバケットに PDF ファイルを追加することで、関数を手動でテストできます。ファイルをソースバケットに追加すると、Lambda 関数が自動的に呼び出され、ファイルの暗号化されたバージョンが送信先のバケットに保存されます。
プロジェクトディレクトリに次のファイルを作成します。
-
test_pdf_encrypt.py- アプリケーションの自動テストに使用できるテストスクリプト -
pytest.ini- テストスクリプトの設定ファイル
以下のセクションを展開してコードを表示し、各ファイルの役割の詳細を確認してください。
次のコードをコピーし、test_pdf_encrypt.py という名前のファイルに貼り付けます。プレースホルダーバケット名は必ず置き換えてください。
-
test_source_bucket_available関数で、amzn-s3-demo-bucketをソースバケットの名前に置き換えます。 -
test_encrypted_file_in_bucket関数で、amzn-s3-demo-bucket-encryptedをsource-bucket-encryptedに置き換えます。source-bucket>はソースバケットの名前です。 -
cleanup関数で、amzn-s3-demo-bucketをソースバケットの名前に置き換え、amzn-s3-demo-bucket-encryptedを送信先バケットの名前に置き換えます。
import boto3 import json import pytest import time import os @pytest.fixture def lambda_client(): return boto3.client('lambda') @pytest.fixture def s3_client(): return boto3.client('s3') @pytest.fixture def logs_client(): return boto3.client('logs') @pytest.fixture(scope='session') def cleanup(): # Create a new S3 client for cleanup s3_client = boto3.client('s3') yield # Cleanup code will be executed after all tests have finished # Delete test.pdf from the source bucket source_bucket = 'amzn-s3-demo-bucket' source_file_key = 'test.pdf' s3_client.delete_object(Bucket=source_bucket, Key=source_file_key) print(f"\nDeleted {source_file_key} from {source_bucket}") # Delete test_encrypted.pdf from the destination bucket destination_bucket = 'amzn-s3-demo-bucket-encrypted' destination_file_key = 'test_encrypted.pdf' s3_client.delete_object(Bucket=destination_bucket, Key=destination_file_key) print(f"Deleted {destination_file_key} from {destination_bucket}") @pytest.mark.order(1) def test_source_bucket_available(s3_client): s3_bucket_name = 'amzn-s3-demo-bucket' file_name = 'test.pdf' file_path = os.path.join(os.path.dirname(__file__), file_name) file_uploaded = False try: s3_client.upload_file(file_path, s3_bucket_name, file_name) file_uploaded = True except: print("Error: couldn't upload file") assert file_uploaded, "Could not upload file to S3 bucket" @pytest.mark.order(2) def test_lambda_invoked(logs_client): # Wait for a few seconds to make sure the logs are available time.sleep(5) # Get the latest log stream for the specified log group log_streams = logs_client.describe_log_streams( logGroupName='/aws/lambda/EncryptPDF', orderBy='LastEventTime', descending=True, limit=1 ) latest_log_stream_name = log_streams['logStreams'][0]['logStreamName'] # Retrieve the log events from the latest log stream log_events = logs_client.get_log_events( logGroupName='/aws/lambda/EncryptPDF', logStreamName=latest_log_stream_name ) success_found = False for event in log_events['events']: message = json.loads(event['message']) status = message.get('record', {}).get('status') if status == 'success': success_found = True break assert success_found, "Lambda function execution did not report 'success' status in logs." @pytest.mark.order(3) def test_encrypted_file_in_bucket(s3_client): # Specify the destination S3 bucket and the expected converted file key destination_bucket = 'amzn-s3-demo-bucket-encrypted' converted_file_key = 'test_encrypted.pdf' try: # Attempt to retrieve the metadata of the converted file from the destination S3 bucket s3_client.head_object(Bucket=destination_bucket, Key=converted_file_key) except s3_client.exceptions.ClientError as e: # If the file is not found, the test will fail pytest.fail(f"Converted file '{converted_file_key}' not found in the destination bucket: {str(e)}") def test_cleanup(cleanup): # This test uses the cleanup fixture and will be executed last pass
自動テストスクリプトは、3 つのテスト関数を実行して、アプリの正しいオペレーションを確認します。
-
テスト
test_source_bucket_availableでは、テスト PDF ファイルをバケットにアップロードすることで、ソースバケットが正常に作成されたことを確認します。 -
テスト
test_lambda_invokedは、関数の最新の CloudWatch Logs ログストリームを調べて、テストファイルをアップロードしたときに Lambda 関数が実行され、成功が報告されていることを確認します。 -
テスト
test_encrypted_file_in_bucketでは、レプリケート先バケットに暗号化されたtest_encrypted.pdfファイルが含まれていることを確認します。
これらのテストがすべて実行されると、スクリプトは追加のクリーンアップステップを実行して、送信元バケットと送信先バケットの両方から test.pdf および test_encrypted.pdf ファイルを削除します。
AWS SAM テンプレートと同様に、このファイルで指定されたバケット名はプレースホルダーです。テストを実行する前に、このファイルをアプリケーションの実際のバケット名で編集する必要があります。このステップについては、「自動スクリプトを使用したアプリケーションのテスト」で詳しく説明します。
次のコードをコピーし、pytest.ini という名前のファイルに貼り付けます。
[pytest] markers = order: specify test execution order
これは、test_pdf_encrypt.py スクリプト内のテストを実行する順序を指定するために必要です。
テストを実行するには、次の手順を実行します。
-
pytestモジュールがローカル環境にインストールされていることを確認します。pytestは以下のコマンドを実行することでインストールできます。pip install pytest -
test.pdfという名前の PDF ファイルを、test_pdf_encrypt.pyファイルとpytest.iniファイルがあるディレクトリに保存します。 -
ターミナルまたはシェルプログラムを開き、テストファイルがあるディレクトリから次のコマンドを実行します。
pytest -s -vテストが完了すると、次のようになります。
============================================================== test session starts ========================================================= platform linux -- Python 3.12.2, pytest-7.2.2, pluggy-1.0.0 -- /usr/bin/python3 cachedir: .pytest_cache hypothesis profile 'default' -> database=DirectoryBasedExampleDatabase('/home/pdf_encrypt_app/.hypothesis/examples') Test order randomisation NOT enabled. Enable with --random-order or --random-order-bucket=<bucket_type> rootdir: /home/pdf_encrypt_app, configfile: pytest.ini plugins: anyio-3.7.1, hypothesis-6.70.0, localserver-0.7.1, random-order-1.1.0 collected 4 items test_pdf_encrypt.py::test_source_bucket_available PASSED test_pdf_encrypt.py::test_lambda_invoked PASSED test_pdf_encrypt.py::test_encrypted_file_in_bucket PASSED test_pdf_encrypt.py::test_cleanup PASSED Deleted test.pdf from amzn-s3-demo-bucket Deleted test_encrypted.pdf from amzn-s3-demo-bucket-encrypted =============================================================== 4 passed in 7.32s ==========================================================
次のステップ
これで、サンプルアプリケーションを作成できました。このコードをベースにして、他の種類のファイル処理アプリケーションを作成することができます。lambda_function.py ファイル内のコードを変更して、ユースケースのファイル処理ロジックを実装します。
一般的なファイル処理のユースケースの多くには、画像処理が含まれます。Python で画像処理を行う場合、Pillow
AWS SAM を使用してリソースをデプロイする場合、デプロイパッケージに適切なソースディストリビューションを含めるための追加の手順を実行する必要があります。AWS SAM はビルドマシンとは異なるプラットフォームの依存関係をインストールしないため、ビルドマシンが Lambda 実行環境とは異なるオペレーティングシステムまたはアーキテクチャを使用している場合、requirements.txt ファイルに正しいソースディストリビューション (.whl ファイル) を指定しても機能しません。代わりに、次のいずれかの操作を行います。
-
sam buildを実行するときは、--use-containerオプションを使用します。このオプションを指定すると、AWS SAM は Lambda 実行環境と互換性のあるコンテナベースイメージをダウンロードし、そのイメージを使用して関数のデプロイパッケージを Docker コンテナに構築します。詳細については、「指定されたコンテナ内における Lambda 関数の構築」を参照してください。 -
正しいソースディストリビューションバイナリを使用して関数の .zip デプロイパッケージを自分で構築し、AWS SAM テンプレートの
CodeUriとして指定したディレクトリに .zip ファイルを保存します。バイナリディストリビューションを使用して Python 用の .zip デプロイパッケージを構築する方法の詳細については、「依存関係を含めて .zip デプロイパッケージを作成する」および「ネイティブライブラリとともに .zip デプロイパッケージを作成する」を参照してください。
