Skip to content

Add @event_soruce decorator for constructing event source data classes #19

Closed
@michaelbrewer

Description

@michaelbrewer

Runtime:
Python (but could apply to all languages that support decorators and lack existing libraries for event source data classes)

Is your feature request related to a problem? Please describe

Originally raises in the following issue: aws-powertools/powertools-lambda-python#434

Constructing a data class from an existing event could be cleaner, and tools like MyPy would complain about change the type of event

def lambda_handler(event: Dict[str, Any], context: LambdaContext) -> Dict[str, Any]:
    event: APIGatewayProxyEventV2 = APIGatewayProxyEventV2(event)
    ...

and the more version would mean an floating event variable:

```python
def lambda_handler(_event: Dict[str, Any], context: LambdaContext) -> Dict[str, Any]:
    event = APIGatewayProxyEventV2(_event)
    ...

Describe the solution you'd like

Creating a thin decorator like event_source which just constructors an instance of the passed in class type would fix this.

@event_source(data_class=APIGatewayProxyEventV2)
def lambda_handler(event: APIGatewayProxyEventV2, context: LambdaContext) -> Dict[str, Any]:
   ...

Describe alternatives you've considered

N/A

If you provide guidance, is this something you'd like to contribute?

Implementation see: aws-powertools/powertools-lambda-python#442

Additional context

Note, to make this work in combination with the idempotent decorator we should allow for the generation of the idempotent key when combined with the @event_source decorator, otherwise there will be a type error when trying to generate the json. Example code which would raise a TypeError:

persistence_layer = DynamoDBPersistenceLayer(table_name="IdempotencyTable")

@event_source(data_class=APIGatewayProxyEventV2)
@idempotent(persistence_store=persistence_layer)
def lambda_handler(event: APIGatewayProxyEventV2, context):
    assert isinstance(event, APIGatewayProxyEventV2)
    ...

Fortunately this would be a simple fix:

        if hasattr(data, "raw_event"):
            data = data.raw_event
        hashed_data = self.hash_function(json.dumps(data, cls=Encoder).encode())

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions