Feature request: Parser support from multiple event sources using the same Lambda

[leandrodamascena]

Use case

Context: AWS Lambda supports multiple different event sources mapping for the same Lambda - https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html

I'm opening this issue to discuss if the Lambda Powertools can support different parsers in a Lambda that has different EventSources. This thread started in our Discord server and it's interesting to hear customer feedback on this.

Original discussion: https://discord.com/channels/1006478942305263677/1006527338621710376/1103290540226785342

Solution/User Experience

We need to investigate this.

Alternative solutions

No response

Acknowledgment

• This feature request meets Lambda Powertools Tenets
• Should this be considered in other Lambda Powertools languages? i.e. Java, TypeScript, and .NET

[rubenfonseca]

Does this mean that the same Lambda would be connected to different event sources?

[leandrodamascena]

Does this mean that the same Lambda would be connected to different event sources?

Exactly man! In my point of view, a single lambda should only support a single event source because you can keep the single responsibility for that lambda and focus your business logic on it. But I see more and more users using fat lambdas, and Lambda supports multiple event sources, so I think we must hear from customers what the use case is and if the Powertools can help them.

What do you think about this Ruben?

[rubenfonseca]

This would be horrible in Go :)

The thing I'm worried about is that you would have to have a clear identifier mark for each kind of even source, so you should be able to distinguish SQS/SNS/etc from a generic dictionary... that isn't fun isn't it? So it would work like a "reverse-parser": here's a dict, tell me what kind of payload it is. Really not sure how to feel about this...

[leandrodamascena]

that isn't fun isn't it? So it would work like a "reverse-parser": here's a dict, tell me what kind of payload it is. Really not sure how to feel about this...

I thought the same, but it seems difficult and non-performatic to do this.

[JeffDGAWS]

I've done this before in Python...it works, but the code needs to understand what type of event it's handling and parse it appropriately. I don't want to imagine the pain-in-the-backside this would be in a staticly typed languge like Go or .NET.

[am29d]

Does it mean the message within SQS and SNS will have the same structure and we have to unwrap the body nicely with one signature without additional conditional expressions?

I think with 3.10 pydantic can handle these type of situations:

from typing import List, Union
import pytest
from pydantic import BaseModel

   
class SQSRecord(BaseModel):
    eventSource: str #"aws:sqs"
    awsRegion: str
    messageId: str
    body: str


class SNSPayload(BaseModel):
    Timestamp: str
    MessageId: str
    Message: str


class SNSRecord(BaseModel):
    EventSource: str #"aws:sns"
    EventVersion: str
    Sns: SNSPayload

class Events(BaseModel):
    Records: Union[List[SQSRecord],  List[SNSRecord]]

def lambda_handler(event, context):
    match Events(**event).Records[0]:
        case SNSRecord(EventSource="aws:sns"):
            return handle_sns(event, context)
        case SQSRecord(eventSource="aws:sqs"):
            return handle_sqs(event, context)
        case _:
            print("unknown event, you probably don't know what you are doing.")


def handle_sns(event, contex):
    print('handling sns event')
    return Events(**event).Records[0].Sns.Message


def handle_sqs(event, contex):
    print('handling sqs event')
    return Events(**event).Records[0].body


@pytest.fixture()
def sns_event():
    """ Generates SNS Event"""

    return {
        "Records": [
            {
                "EventVersion": "1.0",
                "EventSource": "aws:sns",
                "Sns": {
                    "Timestamp": "2019-01-02T12:45:07.000Z",
                    "MessageId": "95df01b4-ee98-5cb9-9903-4c221d41eb5e",
                    "Message": "Hello from SNS!",
                }
            }
        ]
    }

@pytest.fixture()
def sqs_event():
    """ Generates SQS Event"""
    return {
        "Records": [
            {
                "eventVersion": "2.0",
                "eventSource": "aws:sqs",
                "awsRegion": "us-east-1",
                "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
                "body": "Hello from SQS!"
            }
        ]
    }


def test_lambda_handler(sns_event, sqs_event):
    ret = lambda_handler(sns_event, "")
    assert ret == "Hello from SNS!"

    ret = lambda_handler(sqs_event, "")
    assert ret == "Hello from SQS!"

Since we already have built-in models we can add this use case to our docs.

[ran-isenberg]

Does it mean the message within SQS and SNS will have the same structure and we have to unwrap the body nicely with one signature without additional conditional expressions?

I think with 3.10 pydantic can handle these type of situations:

from typing import List, Union
import pytest
from pydantic import BaseModel

   
class SQSRecord(BaseModel):
    eventSource: str #"aws:sqs"
    awsRegion: str
    messageId: str
    body: str


class SNSPayload(BaseModel):
    Timestamp: str
    MessageId: str
    Message: str


class SNSRecord(BaseModel):
    EventSource: str #"aws:sns"
    EventVersion: str
    Sns: SNSPayload

class Events(BaseModel):
    Records: Union[List[SQSRecord],  List[SNSRecord]]

def lambda_handler(event, context):
    match Events(**event).Records[0]:
        case SNSRecord(EventSource="aws:sns"):
            return handle_sns(event, context)
        case SQSRecord(eventSource="aws:sqs"):
            return handle_sqs(event, context)
        case _:
            print("unknown event, you probably don't know what you are doing.")


def handle_sns(event, contex):
    print('handling sns event')
    return Events(**event).Records[0].Sns.Message


def handle_sqs(event, contex):
    print('handling sqs event')
    return Events(**event).Records[0].body


@pytest.fixture()
def sns_event():
    """ Generates SNS Event"""

    return {
        "Records": [
            {
                "EventVersion": "1.0",
                "EventSource": "aws:sns",
                "Sns": {
                    "Timestamp": "2019-01-02T12:45:07.000Z",
                    "MessageId": "95df01b4-ee98-5cb9-9903-4c221d41eb5e",
                    "Message": "Hello from SNS!",
                }
            }
        ]
    }

@pytest.fixture()
def sqs_event():
    """ Generates SQS Event"""
    return {
        "Records": [
            {
                "eventVersion": "2.0",
                "eventSource": "aws:sqs",
                "awsRegion": "us-east-1",
                "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
                "body": "Hello from SQS!"
            }
        ]
    }


def test_lambda_handler(sns_event, sqs_event):
    ret = lambda_handler(sns_event, "")
    assert ret == "Hello from SNS!"

    ret = lambda_handler(sqs_event, "")
    assert ret == "Hello from SQS!"

Since we already have built-in models we can add this use case to our docs.

I've worked with union before with pydantic and it has some issues when two schemas share common fields (results in invalid parsing/exceptions but it might be fixed).

I think it's really important to understand why someone would connect one lambda to multiple sources and decide whether it's something that AWS/community should recommend doing.

Usually, different outputs come from different pipelines, require different permissions and they usually process the records differently.
Having one lambda handle both pipelines mean that:

1. If you connect an SQS with API GW (i've seen people do it), now the return value of the parse function turns into a Union of list of records and a single item. seems quite awkward.
2. Two types of error handling
3. Multiple permissions/ subscription
4. Very hard to differentiate between the pipelines in regards to total invocation time, memory usage etc when looking at cloudwatch metrics.
5. Looking at specific pipeline logs gets complicated as the same log group now shares both log group. It can be sorted but you need to be conscious of it (adding an identifier to all logs such as 'source=sns' etc.).
6. Testing is more complicated in regards to error handling.
7. in many cases, different pipeliens also require transforming the data, not just getting it from the input, so now your lambda handles two responsibilities and handles two types of logic. Is it good? I'm not sure. I'd rather split it and "pay" for an extra function and deploy time.

Can it be done technically? yes
Like @am29d said, it's already possible to achieve it with regular pydantic and unions but i dont think it's something i'd advocate for.

Should you? in my opinion, no but there's no 100% truth here - it's all about pros/cons but again, since it can achieved today with parser, i think it should be left alone.

[heitorlessa]

so where did we land here @leandrodamascena ?