trove-lambda

<!-- AUTO-GENERATED from SKILL.md.tmpl — do not edit directly --> <!-- Regenerate: bun run build:skills -->

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

def parse_sns_direct(event: dict) -> list[dict]:
    # A Lambda subscribed to the topic directly gets no `body`.
    return [json.loads(record["Sns"]["Message"]) for record in event["Records"]]

Check which shape the trigger delivers before parsing. SNS through SQS puts the SNS envelope in body. A direct subscription puts it under Sns.

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)
• Keep messages static and put values in extra: logger.warning("IPC token rejected", extra={"source": source})

IPC Pattern (Backend Communication)

Inbound: an event that claims to come from the backend is untrusted until its token matches BACKEND_IPC_TOKEN. Check it before parsing the payload or doing any work, compare in constant time, and never log the token itself.

import hmac

def ipc_token_valid(presented: str | None) -> bool:
    expected = os.environ["BACKEND_IPC_TOKEN"]  # injected from Secrets Manager
    return presented is not None and hmac.compare_digest(presented, expected)

An HTTP-triggered handler reads it from the Authorization header and returns 401 on a mismatch. A queued event carries it where the backend put it, such as an SNS message attribute, and a mismatch is logged and dropped, not retried.

Outbound:

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

AI Gotchas

• Cold starts: Keep handler imports minimal; use lazy initialization
• Timeout: Default 3s is too low — set explicitly (30s for API, 300s for batch)
• Event source: Always validate event structure before processing
• Idempotency: SQS may deliver messages more than once — design for at-least-once