Lambda使っていて、ふとLambdaから他のLambda関数って呼べないのかなと思って調べてみたらできたのでその時のメモ。

InvocationTypeで同期処理か非同期処理かの呼び出し方法を指定できる。期処理がよければRequestResponseを指定、非同期ならEventを指定する。非同期処理にすると最初のLambdaはすぐにレスポンスを返してくれるので、Lambda呼び出し先のタイムアウト等を気にする必要がなくなる。呼び出した関数からのレスポンスを使いたいば同期処理にすればよい。

同期で関数を呼びたい場合

import os
import json
import boto3
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 呼び出したい関数名
function_name = os.environ.get('LAMBDA_FUNCTION_NAME')

def lambda_handler(event: dict, context):
    # ログ出力
    logging.info(json.dumps(event))

    client = boto3.client('lambda')
    response = client.invoke(
        FunctionName=function_name,
        InvocationType='RequestResponse',
        LogType='Tail',
        Payload=json.dumps(event)
    )

    return {
        'statusCode': 200,
        'body': json.dumps('OK')
    }

非同期で関数を呼びたい場合

import os
import json
import boto3
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 呼び出したい関数名
function_name = os.environ.get('LAMBDA_FUNCTION_NAME')

def lambda_handler(event: dict, context):
    # ログ出力
    logging.info(json.dumps(event))

    client = boto3.client('lambda')
    client.invoke(
        FunctionName=function_name,
        InvocationType='Event',
        LogType='Tail',
        Payload=json.dumps(event)
    )

    return {
        'statusCode': 200,
        'body': json.dumps('OK')
    }

やっていてハマったのが、非同期処理にしていても以下の用にレスポンスを変数で受け取るようにすると結果が帰ってくるまで最初のLambdaの実行が終わらない。そのため、場合によっては呼び出し元がタイムアウトと勘違いをしてリトライしてくる可能性がある。 SlackからLambdaのAPIをたたいていて実際にあった。Slackは3秒以内にレスポンスが帰らなければリトライするようです。

    response = client.invoke(
        FunctionName=function_name,
        InvocationType='Event',
        LogType='Tail',
        Payload=json.dumps(event)
    )

ちょっと長い処理だったり、条件によって処理をわけたい場合等に使うと力を発揮してくれそう。個人的には非同期処理がLambdaだけでできるのは助かるな。