AWS lambda関数の作成から実行まで

AWS Lambda関数を作成し、起動確認するまでの手順を解説します。
※ 操作はすべてAWS Python ライブラリ(boto3)を使用して実施します。

以降の内容は以下の記事を前提とします

• AWS organization を利用して新しいAWS環境をゼロから作成する

ロールへの権限付与

Lambda関数実行に必要な権限をロールに付与します。

import boto3
from botocore.exceptions import ClientError

# Configuration constants
PROFILE_NAME = 'resources_dev_admin'
ROLE_NAME = 'ResourcesEnvMainRole'

# List of IAM policies to attach
POLICY_ARNS = [
    'arn:aws:iam::aws:policy/AWSLambda_FullAccess',
    'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole',
    'arn:aws:iam::aws:policy/CloudWatchLogsFullAccess'
]

# Initialize AWS session and IAM client
session = boto3.Session(profile_name=PROFILE_NAME)
iam_client = session.client('iam')

print(f"Starting to attach policies to role '{ROLE_NAME}' using profile '{PROFILE_NAME}'...")

# Loop through each policy and attach it to the role
for policy_arn in POLICY_ARNS:
    try:
        iam_client.attach_role_policy(
            RoleName=ROLE_NAME,
            PolicyArn=policy_arn
        )
        print(f"Success: Attached {policy_arn.split('/')[-1]}.")
    except ClientError as e:
        print(f"Error: Failed to attach {policy_arn.split('/')[-1]}.")
        print(e.response['Error']['Message'])

print("Process completed.")

lambda関数の作成

import os
import zipfile
import boto3

PROFILE_NAME = 'resources_dev'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'


PYTHON_FILE_NAME = "lambda_function.py"
ZIP_FILE_NAME = "lambda_function.zip"

content = """
import json

def lambda_handler(event, context):
    print("lambda_function#lambda_handler called...")
    print("Received event: " + json.dumps(event, indent=2))
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Python Lambda')
    }
"""

with open(PYTHON_FILE_NAME, "w", encoding="utf-8") as f:
    f.write(content)

with zipfile.ZipFile(ZIP_FILE_NAME, "w", zipfile.ZIP_DEFLATED) as zipf:
    zipf.write(PYTHON_FILE_NAME)

if os.path.exists(PYTHON_FILE_NAME):
    os.remove(PYTHON_FILE_NAME)


session = boto3.Session(profile_name=PROFILE_NAME)

iam_client = session.client('iam')
role_response = iam_client.get_role(RoleName=ROLE_NAME)
ROLE_ARN = role_response['Role']['Arn']

lambda_client = session.client('lambda')

with open('lambda_function.zip', 'rb') as f:
    zip_data = f.read()

response = lambda_client.create_function(
    FunctionName=FUNCTION_NAME,
    Runtime='python3.14',
    Role=ROLE_ARN,
    Handler='lambda_function.lambda_handler',
    Code={
        'ZipFile': zip_data
    },
    Timeout=180
)

print(f"Function {response['FunctionName']} created successfully.")

if os.path.exists(ZIP_FILE_NAME):
    os.remove(ZIP_FILE_NAME)

lambda関数の実行

import json
import boto3

PROFILE_NAME = 'resources_dev'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'

session = boto3.Session(
    profile_name=PROFILE_NAME
)
lambda_client = session.client('lambda')

payload = {
    "key1": "value1",
    "key2": "value2"
}

try:
    response = lambda_client.invoke(
        FunctionName=FUNCTION_NAME,
        InvocationType='RequestResponse',
        Payload=json.dumps(payload)
    )
    
    response_payload = json.loads(response['Payload'].read().decode('utf-8'))
    print("success:", response_payload)

except Exception as e:
    print("error:", e)

lambda関数実行ログの確認

from datetime import datetime, timedelta, timezone
import boto3

PROFILE_NAME = 'resources_dev'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'

LOG_GROUP_NAME = f"/aws/lambda/{FUNCTION_NAME}"

session = boto3.Session(profile_name=PROFILE_NAME)
client = session.client("logs")

SINCE_MINUTES = 10
JST = timezone(timedelta(hours=9))  # JST (UTC+9)

start_time = int(
    (datetime.now(JST) - timedelta(minutes=SINCE_MINUTES)).timestamp() * 1000
)

paginator = client.get_paginator("filter_log_events")

for page in paginator.paginate(
    logGroupName=LOG_GROUP_NAME, startTime=start_time
):
    for event in page.get("events", []):
        dt = datetime.fromtimestamp(event["timestamp"] / 1000, JST)
        print(f"[{dt.isoformat()}] {event['message'].rstrip()}")

ログ保持期間の設定

import boto3

PROFILE_NAME = 'resources_dev'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'
RETENTION     = 3

LOG_GROUP = f"/aws/lambda/{FUNCTION_NAME}"

session = boto3.Session(
    profile_name=PROFILE_NAME
)
client = session.client('logs')

response = client.put_retention_policy(
    logGroupName=LOG_GROUP,
    retentionInDays=RETENTION
)

print(f"Successfully set retention period for '{LOG_GROUP}' to {RETENTION} days.")

lambda関数一括作成(ロール設定/関数作成/実行/ログ保持期間設定)

lambda関数を一括作成します(ロール設定/関数作成/実行/ログ保持期間設定)

import json
import os
import zipfile
import boto3
import time
from botocore.exceptions import ClientError
from datetime import datetime, timedelta, timezone

ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'


# Configuration constants
PROFILE_NAME_ADMIN = 'resources_dev_admin'

# List of IAM policies to attach
POLICY_ARNS = [
    'arn:aws:iam::aws:policy/AWSLambda_FullAccess',
    'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole',
    'arn:aws:iam::aws:policy/CloudWatchLogsFullAccess'
]

# Initialize AWS session and IAM client
session = boto3.Session(profile_name=PROFILE_NAME_ADMIN)
iam_client = session.client('iam')

print(f"Starting to attach policies to role '{ROLE_NAME}' using profile '{PROFILE_NAME_ADMIN}'...")

# Loop through each policy and attach it to the role
for policy_arn in POLICY_ARNS:
    try:
        iam_client.attach_role_policy(
            RoleName=ROLE_NAME,
            PolicyArn=policy_arn
        )
        print(f"Success: Attached {policy_arn.split('/')[-1]}.")
    except ClientError as e:
        print(f"Error: Failed to attach {policy_arn.split('/')[-1]}.")
        print(e.response['Error']['Message'])

print("Process completed.")


PYTHON_FILE_NAME = "lambda_function.py"
ZIP_FILE_NAME = "lambda_function.zip"

content = """
import json

def lambda_handler(event, context):
    print("lambda_function#lambda_handler called...")
    print("Received event: " + json.dumps(event, indent=2))
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Python Lambda')
    }
"""

with open(PYTHON_FILE_NAME, "w", encoding="utf-8") as f:
    f.write(content)

with zipfile.ZipFile(ZIP_FILE_NAME, "w", zipfile.ZIP_DEFLATED) as zipf:
    zipf.write(PYTHON_FILE_NAME)

if os.path.exists(PYTHON_FILE_NAME):
    os.remove(PYTHON_FILE_NAME)


# Use the same admin session for subsequent AWS operations
iam_client = session.client('iam')
role_response = iam_client.get_role(RoleName=ROLE_NAME)
ROLE_ARN = role_response['Role']['Arn']

lambda_client = session.client('lambda')

with open('lambda_function.zip', 'rb') as f:
    zip_data = f.read()

response = lambda_client.create_function(
    FunctionName=FUNCTION_NAME,
    Runtime='python3.14',
    Role=ROLE_ARN,
    Handler='lambda_function.lambda_handler',
    Code={
        'ZipFile': zip_data
    },
    Timeout=180
)

print(f"Function {response['FunctionName']} created successfully.")


if os.path.exists(ZIP_FILE_NAME):
    os.remove(ZIP_FILE_NAME)


# Wait automatically until the Lambda function status becomes Active
print("Waiting for the Lambda function to become Active...")
waiter = lambda_client.get_waiter('function_active_v2')
waiter.wait(
    FunctionName=FUNCTION_NAME,
    WaiterConfig={'Delay': 2, 'MaxAttempts': 30}  # Check every 2 seconds, up to 30 times (60 seconds max)
)
print("Lambda function is now Active.")


payload = {
    "key1": "value1",
    "key2": "value2"
}

try:
    response = lambda_client.invoke(
        FunctionName=FUNCTION_NAME,
        InvocationType='RequestResponse',
        Payload=json.dumps(payload)
    )
    
    response_payload = json.loads(response['Payload'].read().decode('utf-8'))
    print("1st call successed:", response_payload)

except Exception as e:
    print("error:", e)

print('wait 10s')
time.sleep(10)

LOG_GROUP_NAME = f"/aws/lambda/{FUNCTION_NAME}"

log_client = session.client("logs")

SINCE_MINUTES = 10
JST = timezone(timedelta(hours=9))  # JST (UTC+9)

start_time = int(
    (datetime.now(JST) - timedelta(minutes=SINCE_MINUTES)).timestamp() * 1000
)

paginator = log_client.get_paginator("filter_log_events")

for page in paginator.paginate(
    logGroupName=LOG_GROUP_NAME, startTime=start_time
):
    for event in page.get("events", []):
        dt = datetime.fromtimestamp(event["timestamp"] / 1000, JST)
        print(f"[{dt.isoformat()}] {event['message'].rstrip()}")


RETENTION     = 3

response = log_client.put_retention_policy(
    logGroupName=LOG_GROUP_NAME,
    retentionInDays=RETENTION
)

print(f"Successfully set retention period for '{LOG_GROUP_NAME}' to {RETENTION} days.")

lambda関数の削除（構成が不要になった場合のみ）

import boto3

PROFILE_NAME = 'resources_dev'
FUNCTION_NAME = 'func_test_01'

session = boto3.Session(
    profile_name=PROFILE_NAME
)

lambda_client = session.client('lambda')

try:
    deleted_count = 0
    paginator = lambda_client.get_paginator('list_event_source_mappings')
    
    for page in paginator.paginate(FunctionName=FUNCTION_NAME):
        for mapping in page.get('EventSourceMappings', []):
            uuid = mapping['UUID']
            arn = mapping['EventSourceArn']
            print(f"Deleting event source mapping: {uuid} ({arn})")
            
            lambda_client.delete_event_source_mapping(UUID=uuid)
            deleted_count += 1

    if deleted_count == 0:
        print(f"No event source mappings found for '{FUNCTION_NAME}'. Skipped.")
    else:
        print(f"Successfully triggered deletion for {deleted_count} mapping(s).")

except lambda_client.exceptions.ResourceNotFoundException:
    print(f"Lambda function '{FUNCTION_NAME}' does not exist. Skipped.")
except Exception as e:
    print(f"Error deleting event source mappings: {e}")

response = lambda_client.delete_function(
    FunctionName=FUNCTION_NAME
)

print(response)

LOG_GROUP_NAME = f"/aws/lambda/{FUNCTION_NAME}"

log_client = session.client('logs')
response = log_client.delete_log_group(
    logGroupName=LOG_GROUP_NAME
)

関連記事

• AWS organization を利用して新しいAWS環境をゼロから作成する
• AWS lambda関数の作成から実行まで
• AWS lambda関数と SQS SNS との連携
• AWS APIGateway からの SNS 連携
• AWS APIGateway からの lambda 連携
• AWS Lambda のソースを更新する
• AWS Lambda でLine bot を作成する
• AWS Lambda で Telegram bot を作成する

AWS lambda関数の作成から実行まで

更新日:2026年08月15日