Function get_event_source

zappa/utilities.py:710–737

Given an event_source dictionary item, a session and a lambda_arn, create and return the appropriate event source object.

(
    event_source: Dict[str, Any], lambda_arn: str, target_function: str, boto_session: boto3.Session, dry: bool = False
)

def get_event_source(
    event_source: Dict[str, Any], lambda_arn: str, target_function: str, boto_session: boto3.Session, dry: bool = False
) -> Tuple[BaseEventSource, str]:
    """
    Given an event_source dictionary item, a session and a lambda_arn,
    create and return the appropriate event source object.
    """
    event_source_map: Dict[str, type[BaseEventSource]] = {
        "dynamodb": DynamoDBStreamEventSource,
        "kinesis": KinesisEventSource,
        "s3": S3EventSource,
        "sns": SNSEventSource,
        "sqs": SqsEventSource,
        "events": CloudWatchEventSource,
    }

    arn = event_source["arn"]
    _, _, svc, _ = arn.split(":", 3)

    event_source_class = event_source_map.get(svc, None)
    if not event_source_class:
        raise ValueError("Unknown event source: {0}".format(arn))

    function_arn = lambda_arn

    event_source_obj = event_source_class(boto_session, event_source)

    return event_source_obj, function_arn


def add_event_source(