0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@saurus12(サウルス１２)

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

0
Posted at

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

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

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

ロールへの権限付与

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 lambda関数の作成から実行まで

更新日:2026年08月15日

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?