trove-lambda

Trove · v2026.7.4

Session Init

This skill ships Trove conventions. Prefer existing project patterns over generic best practices when they conflict.

AWS Lambda Conventions

Handler Structure

from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()

@logger.inject_lambda_context(log_event=True)
def handler(event: dict, context: LambdaContext) -> dict:
    records = parse_event(event)
    for record in records:
        process_record(record)
    return {"statusCode": 200}

SQS/SNS Event Parsing

def parse_sqs_event(event: dict) -> list[dict]:
    return [json.loads(record["body"]) for record in event["Records"]]

def parse_sns_via_sqs(event: dict) -> list[dict]:
    messages = []
    for record in event["Records"]:
        body = json.loads(record["body"])
        message = json.loads(body["Message"])
        messages.append(message)
    return messages

Logging

Follow the same logging rules as trove-python, plus:

• Use @logger.inject_lambda_context(log_event=True) on handlers
• Use logger.append_keys() for request-scoped context
• Use logger.exception() in error handlers (captures stack trace for Sentry)

IPC Pattern (Backend Communication)

async def call_backend(endpoint: str, payload: dict) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BACKEND_URL}/api/ipc/{endpoint}",
            json=payload,
            headers={"Authorization": f"Bearer {IPC_TOKEN}"},
            timeout=30.0,
        )
        response.raise_for_status()
        return response.json()

CDK Patterns

• One construct per Lambda function
• Use PythonFunction from @aws-cdk/aws-lambda-python-alpha
• Environment variables via CDK environment prop, secrets via Secrets Manager
• Set memorySize, timeout, and reservedConcurrentExecutions