AWS APIGateway からの SNS 連携

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

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

• AWS lambda関数と SQS SNS との連携

ロールのポリシー更新

ロールをAPIGatewayに適用できるようポリシー設定を更新します

import json
import boto3

ROLE_NAME = 'ResourcesEnvMainRole'
AWS_PROFILE = "resources_dev_admin"

session1 = boto3.Session()

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

session2 = boto3.Session(profile_name=AWS_PROFILE)
client = session2.client('iam')

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"
        }
    ]
}

response = client.update_assume_role_policy(
    RoleName=ROLE_NAME,
    PolicyDocument=json.dumps(trust_policy),
)

print(response)

ロールへの権限付与

APIGatewayとの連携に必要な権限をロールに付与します。

import boto3
from botocore.exceptions import ClientError

PROFILE_NAME = 'resources_dev_admin'
ROLE_NAME = 'ResourcesEnvMainRole'

POLICY_ARNS = [
    'arn:aws:iam::aws:policy/AmazonAPIGatewayAdministrator'
]

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

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

ロールへのiam:PassRole権限の付与

API Gatewayがバックエンド(SNS)を呼び出すためiam:PassRole権限の付与を行う

import json
import boto3

PROFILE_NAME = 'resources_dev_admin'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'

POLICY_NAME = 'ApiGatewayPassRolePolicy'

session = boto3.Session(profile_name=PROFILE_NAME)
iam_client = session.client('iam')

sts_client = session.client('sts')
aws_account_id = sts_client.get_caller_identity()['Account']
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}")

API Gatewayの作成

import boto3

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

HTTP_METHOD = "POST"
STAGE_NAME = "dev"

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

aws_region = session.region_name
sts_client = session.client('sts')
aws_account_id = sts_client.get_caller_identity()['Account']

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("ルートリソース (/) が見つかりませんでした。")

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

    credentials_arn = f"arn:aws:iam::{aws_account_id}:role/{ROLE_NAME}"
    sns_topic_arn = f"arn:aws:sns:{aws_region}:{aws_account_id}:{TOPIC_NAME}"
    integration_uri = f"arn:aws:apigateway:{aws_region}:sns:path//"

    request_parameters = {
        "integration.request.querystring.Action": "'Publish'",
        "integration.request.querystring.TopicArn": f"'{sns_topic_arn}'",
        "integration.request.querystring.Message": "method.request.body"
    }

    response = apigateway_client.put_integration(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        type='AWS',
        integrationHttpMethod=HTTP_METHOD,
        uri=integration_uri,
        credentials=credentials_arn,
        requestParameters=request_parameters,
        passthroughBehavior='WHEN_NO_TEMPLATES'
    )
    print(f"Successfully integrated {HTTP_METHOD} method with SNS topic: {TOPIC_NAME}")

    apigateway_client.put_method_response(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        statusCode='200'
    )

    response = apigateway_client.put_integration_response(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        statusCode='200',
        selectionPattern=""
    )

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

POST 送信をテストする

import json
import urllib.request

INVOKE_URL = "https://amazonaws.com"

post_data = {
    "message": "Hello from API Gateway to SNS!",
    "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}")

ログの確認

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()}")

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

import json
import boto3
import urllib.request
from botocore.exceptions import ClientError

PROFILE_NAME_ADMIN = 'resources_dev_admin'

PROFILE_NAME = 'resources_dev_admin'
ROLE_NAME = 'ResourcesEnvMainRole'
FUNCTION_NAME = 'func_test_01'
API_NAME = f"{FUNCTION_NAME}-api"
TOPIC_NAME = f"{FUNCTION_NAME}-sns"


session1 = boto3.Session()

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

session2 = boto3.Session(profile_name=PROFILE_NAME_ADMIN)
iam_client = session2.client('iam')

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"
        }
    ]
}

response = iam_client.update_assume_role_policy(
    RoleName=ROLE_NAME,
    PolicyDocument=json.dumps(trust_policy),
)

print(response)


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


POLICY_NAME = 'ApiGatewayPassRolePolicy'

sts_client2 = session2.client('sts')
aws_account_id = sts_client2.get_caller_identity()['Account']
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}")


HTTP_METHOD = "POST"
STAGE_NAME = "v1"

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

aws_region = session.region_name
sts_client = session.client('sts')
aws_account_id = sts_client.get_caller_identity()['Account']

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("ルートリソース (/) が見つかりませんでした。")

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

    credentials_arn = f"arn:aws:iam::{aws_account_id}:role/{ROLE_NAME}"
    sns_topic_arn = f"arn:aws:sns:{aws_region}:{aws_account_id}:{TOPIC_NAME}"
    integration_uri = f"arn:aws:apigateway:{aws_region}:sns:path//"

    request_parameters = {
        "integration.request.querystring.Action": "'Publish'",
        "integration.request.querystring.TopicArn": f"'{sns_topic_arn}'",
        "integration.request.querystring.Message": "method.request.body"
    }

    response = apigateway_client.put_integration(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        type='AWS',
        integrationHttpMethod=HTTP_METHOD,
        uri=integration_uri,
        credentials=credentials_arn,
        requestParameters=request_parameters,
        passthroughBehavior='WHEN_NO_TEMPLATES'
    )
    print(f"Successfully integrated {HTTP_METHOD} method with SNS topic: {TOPIC_NAME}")

    apigateway_client.put_method_response(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        statusCode='200'
    )

    response = apigateway_client.put_integration_response(
        restApiId=rest_api_id,
        resourceId=root_resource_id,
        httpMethod=HTTP_METHOD,
        statusCode='200',
        selectionPattern=""
    )

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

print('---')
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 APIGateway からの SNS 連携

更新日:2026年08月15日