Feature request: Internal Library event-dispatcher

Start by reading the idempotency persistence-layer code around BasePersistenceLayer.save_success and the proposed core.internal_events dispatcher API. Trace existing hook mechanisms to understand integration points and identify where published events and their documented data structures should be defined. A completed implementation would provide subscription and emission behavior, including the proposed idempotency_saved event, with tests covering notification and compatibility with existing hooks.

Use case

There are several use cases where customers require the ability to react to a change within the Powertools library internals. These have been implemented using a concrete hook-style mechanism where a hook function is provided to respond to a condition. These are fixed and become hard-coded mechanisms of providing a simple hook function capability, which slowly bloats the powertools library.

Instead, we could provide customers with a new internal library utility event_dispatch which enables customers to subscribe to known, published events that occur within the Powertools internal library code and enables customers to subscribe their own function handler to respond to the internal event. The events would be coupled with a known data structure, which must be documented to enable customers access to the required data at the time the event occurs and enable them to provide some form of action.

This would make implementing future function hook-style features more straightforward and could refactor existing hooks into this style, to deprecate the more concrete, fixed examples that exist today (there are not too many, IMO).

Solution/User Experience

Powertools would publish known events and also provide detail on the provided data for the event hook.

for example in idempotency utility there coul dbe a mechanism to provide an event when the idempotent record has been saved to the persistent store (and only when the record is saved):

handler Code:

import os
from dataclasses import dataclass, field
from uuid import uuid4

from aws_lambda_powertools.utilities.idempotency import (
    DynamoDBPersistenceLayer,
    idempotent,
)
from aws_lambda_powertools.utilities.typing import LambdaContext

## New part for event-dispatch
from core.internal_events import dispatcher, internal_events 

def handle_idempotency_record_saved(data):
    # Do something with the idempotency record data now it has been saved - maybe store the ID in another place for future reference
  
# Subscribe to the event
dispatcher.subscribe(internal_events.Idempotency_record_saved, handle_idempotency_record_saved)

table = os.getenv("IDEMPOTENCY_TABLE", "")
persistence_layer = DynamoDBPersistenceLayer(table_name=table)


@dataclass
class Payment:
    user_id: str
    product_id: str
    payment_id: str = field(default_factory=lambda: f"{uuid4()}")


class PaymentError(Exception): ...


@idempotent(persistence_store=persistence_layer)
def lambda_handler(event: dict, context: LambdaContext):
    try:
        payment: Payment = create_subscription_payment(event)
        return {
            "payment_id": payment.payment_id,
            "message": "success",
            "statusCode": 200,
        }
    except Exception as exc:
        raise PaymentError(f"Error creating payment {str(exc)}")


def create_subscription_payment(event: dict) -> Payment:
    return Payment(**event)

Within the POwertools library code (in the core Base Idempotency class):

import dispatcher from core.event_dispatch


class BasePersistenceLayer(ABC):

    def save_success(self, data: dict[str, Any], result: dict) -> None:
        """
        Save record of function's execution completing successfully

        Parameters
        ----------
        data: dict[str, Any]
            Payload
        result: dict
            The response from function
        """
        idempotency_key = self._get_hashed_idempotency_key(data=data)
        if idempotency_key is None:
            # If the idempotency key is None, no data will be saved in the Persistence Layer.
            # See: https://github.com/aws-powertools/powertools-lambda-python/issues/2465
            return None

        response_data = json.dumps(result, cls=Encoder, sort_keys=True)

        data_record = DataRecord(
            idempotency_key=idempotency_key,
            status=STATUS_CONSTANTS["COMPLETED"],
            expiry_timestamp=self._get_expiry_timestamp(),
            response_data=response_data,
            payload_hash=self._get_hashed_payload(data=data),
        )
        logger.debug(
            f"Function successfully executed. Saving record to persistence store with "
            f"idempotency key: {data_record.idempotency_key}",
        )
        self._update_record(data_record=data_record)
        self._save_to_cache(data_record=data_record)

        # Notify subscribers of data record being save
        dispatcher.emit('idempotency_saved', data_record)

Alternative solutions

Keep adding more configuration swithces ot enable function hook capability in the library as new use cases arrive.

Acknowledgment

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