AWS APIGateway からの lambda 連携

http post をAPIGatewayを経由して直接lambdaへ連携します

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

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

API Gateway 連携の一括作成(各種権限付与/API Gateway作成/呼び出し実行)

import json
import time
import boto3
import urllib.request
from botocore.exceptions import ClientError
from datetime import datetime, timezone, timedelta

# --- Configuration Settings ---
PROFILE_NAME_ADMIN = 'resources_dev_admin'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'
API_NAME = f"{FUNCTION_NAME}-api"
HTTP_METHOD = "POST"
STAGE_NAME = "v1"


session1 = boto3.Session()

sts_client = session1.client('sts')
AWS_ROOT_ACCOUNT_ID = sts_client.get_caller_identity()['Account']


# Initialize session with the unified profile
session = boto3.Session(profile_name=PROFILE_NAME_ADMIN)

# Initialize AWS service clients
sts_client = session.client('sts')
iam_client = session.client('iam')
apigateway_client = session.client('apigateway')
lambda_client = session.client('lambda')
logs_client = session.client('logs')

# Retrieve AWS environment metadata
aws_region = session.region_name
aws_account_id = sts_client.get_caller_identity()['Account']

# --- Update IAM Assume Role Policy ---
trust_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": f"arn:aws:iam::{AWS_ROOT_ACCOUNT_ID}:root",
                "Service": [
                    "lambda.amazonaws.com",
                    "apigateway.amazonaws.com"
                ]
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

try:
    response = iam_client.update_assume_role_policy(
        RoleName=ROLE_NAME,
        PolicyDocument=json.dumps(trust_policy),
    )
    print("Successfully updated assume role policy.")
except ClientError as e:
    print(f"Error updating assume role policy: {e.response['Error']['Message']}")

# --- Attach Managed Policies ---
POLICY_ARNS = [
    'arn:aws:iam::aws:policy/AmazonAPIGatewayAdministrator'
]

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.")

# --- Apply Inline Policy (PassRole) ---
POLICY_NAME = 'ApiGatewayPassRolePolicy'
role_arn = f"arn:aws:iam::{aws_account_id}:role/{ROLE_NAME}"

policy_dict = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "iam:PassRole",
            "Resource": role_arn
        }
    ]
}

try:
    response = iam_client.put_role_policy(
        RoleName=ROLE_NAME,
        PolicyName=POLICY_NAME,
        PolicyDocument=json.dumps(policy_dict)
    )
    print(f"Successfully applied policy '{POLICY_NAME}' to role '{ROLE_NAME}'.")
except Exception as e:
    print(f"Error applying role policy: {e}")

# --- Build API Gateway and Integrate with Lambda ---
invoke_url = None

try:
    create_response = apigateway_client.create_rest_api(
        name=API_NAME
    )
    rest_api_id = create_response['id']
    print(f"REST_API_ID: {rest_api_id}")

    resources_response = apigateway_client.get_resources(
        restApiId=rest_api_id
    )
    
    root_resource_id = None
    for item in resources_response.get('items', []):
        if item['path'] == '/':
            root_resource_id = item['id']
            break

    if not root_resource_id:
        raise ValueError("Root resource (/) not found.")

    print(f"ROOT_RESOURCE_ID: {root_resource_id}")

    method_response = apigateway_client.put_method(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        authorizationType='NONE'
    )
    print(f"Successfully created {HTTP_METHOD} method.")

    lambda_arn = f"arn:aws:lambda:{aws_region}:{aws_account_id}:function:{FUNCTION_NAME}"
    integration_uri = f"arn:aws:apigateway:{aws_region}:lambda:path/2015-03-31/functions/{lambda_arn}/invocations"

    response = apigateway_client.put_integration(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        type='AWS_PROXY',
        integrationHttpMethod=HTTP_METHOD,
        uri=integration_uri
    )
    print(f"Successfully integrated {HTTP_METHOD} method with Lambda function: {FUNCTION_NAME}")

    try:
        source_arn = f"arn:aws:execute-api:{aws_region}:{aws_account_id}:{rest_api_id}/*/{HTTP_METHOD}/"
        lambda_client.add_permission(
            FunctionName=FUNCTION_NAME,
            StatementId=f"apigateway-invoke-{rest_api_id}",
            Action="lambda:InvokeFunction",
            Principal="apigateway.amazonaws.com",
            SourceArn=source_arn
        )
        print("Successfully added invoke permission to Lambda function.")
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceConflictException':
            print("Invoke permission already exists.")
        else:
            raise e

    response = apigateway_client.create_deployment(
        restApiId=rest_api_id,
        stageName=STAGE_NAME
    )
    
    invoke_url = f"https://{rest_api_id}.execute-api.{aws_region}.amazonaws.com/{STAGE_NAME}/"
    print(f"Successfully deployed API Gateway to stage: {STAGE_NAME}")
    print(f"Invoke URL: {invoke_url}")

except Exception as e:
    print(f"Error managing API Gateway: {e}")

# --- Test API Call ---
if invoke_url:
    post_data = {
        "message": "Hello from API Gateway to Lambda!",
        "data1": "data1",
        "data2": 2
    }
    encoded_data = json.dumps(post_data).encode('utf-8')

    req = urllib.request.Request(
        url=invoke_url,
        data=encoded_data,
        headers={'Content-Type': 'application/json'},
        method='POST'
    )

    try:
        with urllib.request.urlopen(req) as res:
            status_code = res.getcode()
            response_body = res.read().decode('utf-8')
            print(f"Response Status: {status_code}")
            print(f"Response Body: {response_body}")
            
    except urllib.error.HTTPError as e:
        print(f"HTTP Error: {e.code}")
        print(f"Error Body: {e.read().decode('utf-8')}")
    except urllib.error.URLError as e:
        print(f"URL Error: {e.reason}")

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

# --- Fetch CloudWatch Logs ---
LOG_GROUP_NAME = f"/aws/lambda/{FUNCTION_NAME}"
SINCE_MINUTES = 10
JST = timezone(timedelta(hours=9))

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

paginator = logs_client.get_paginator("filter_log_events")

try:
    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()}")
except ClientError as e:
    if e.response['Error']['Code'] == 'ResourceNotFoundException':
        print(f"Log group {LOG_GROUP_NAME} does not exist yet. Please invoke the function first.")
    else:
        print(f"Error fetching logs: {e}")

print('---')
print('---')
print('')
if invoke_url:
    print(f"Invoke URL: {invoke_url}")
print('')
print('')

API Gatewayの削除（構成が不要になった場合のみ）

import boto3

PROFILE_NAME = 'resources_dev'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'
API_NAME = f"{FUNCTION_NAME}-api"

session = boto3.Session(profile_name=PROFILE_NAME)
apigateway_client = session.client('apigateway')

try:
    rest_api_id = None
    paginator = apigateway_client.get_paginator('get_rest_apis')
    for page in paginator.paginate():
        for api in page.get('items', []):
            if api['name'] == API_NAME:
                rest_api_id = api['id']
                break
        if rest_api_id:
            break

    if rest_api_id:
        apigateway_client.delete_rest_api(restApiId=rest_api_id)
        print(f"Successfully deleted API Gateway: {API_NAME}")
    else:
        print(f"API Gateway '{API_NAME}' does not exist. Skipped.")

except Exception as e:
    print(f"Error deleting API Gateway: {e}")

関連記事

• 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 のソースを更新する
• AWS S3の作成からファイルの登録公開まで

AWS APIGateway からの lambda 連携

更新日:2026年08月15日