def handler()

in lca-ai-stack/source/lambda_functions/call_event_stream_processor/lambda_function.py [0:0]

• 25 lines of code
• 1 McCabe index (conditional complexity)

def handler(event: DynamoDBStreamEvent, context: LambdaContext):
    # pylint: disable=unused-argument
    """Lambda handler

    Processes Call events from the DynamoDB stream of the Event Sourcing table
    It manages the mutation of Call state/metadata including transcript and
    sentiment derived from the Call events.
    """
    LOGGER.debug("lambda event", extra={"event": event})
    LOGGER.info("GraphQL endpoint: [%s]", APPSYNC_GRAPHQL_URL)
    LOGGER.info("sentiment analysis enabled: [%s]", IS_SENTIMENT_ANALYSIS_ENABLED)

    record_count = len(event.get("Records", []))  # type: ignore
    METRICS.add_metric(
        name="EventBatchCount",
        unit=MetricUnit.Count,
        value=record_count,
    )
    event_result = EVENT_LOOP.run_until_complete(handle_event(event=event))

    METRICS.add_metric(
        name="EventInsertErrorCount",
        unit=MetricUnit.Count,
        value=event_result.get("event_error_count", 0.0),
    )
    METRICS.add_metric(
        name="EventInsertCount",
        unit=MetricUnit.Count,
        value=event_result.get("event_insert_count", 0.0),
    )

    # Lambda tumbling window state
    incoming_state = event.get("state", {})  # type: ignore
    outgoing_state = event_result.get("state", incoming_state)

    LOGGER.debug("state", extra=dict(incoming_state=incoming_state, outgoing_state=outgoing_state))
    return {"state": outgoing_state}