wshtlib

Lightweight observability library for AWS Lambda and FastAPI. Zero external dependencies.

A focused alternative to aws-powertools — covers structured logging, CloudWatch metrics (EMF), request context propagation, and Lambda handler boilerplate. Nothing more.

Install

pip install wshtlib

FastAPI/Starlette middleware is optional:

pip install wshtlib[fastapi]

Usage

Lambda handler

Two decorators, one per invocation mode.

@bootstrap — synchronous invocations (API Gateway), where the return value is the response:

from wshtlib import bootstrap, get_logger

logger = get_logger("my-service")

@bootstrap
def handler(event, context):
    logger.info("invoked", path=event.get("path"))
    return {"statusCode": 200}

It handles:

• Warming events ("source": "lambda-warming") — returns 200 early
• Context init and structured log enrichment
• Unhandled exceptions — logs error, returns 500

@worker — asynchronous invocations (S3, EventBridge, SQS), where the return value is discarded:

from wshtlib import worker, get_logger

logger = get_logger("my-worker")

@worker
def handler(event, context):
    logger.info("processing", records=len(event["Records"]))

Same context init and structured error logging, but the exception is re-raised rather than swallowed — retries, on_failure destinations, the DLQ, and the Errors metric all depend on Lambda seeing the invocation fail. No warming-event handling.

Structured logging

from wshtlib import get_logger

logger = get_logger("my-service")
logger.info("user signed in", user_id="u_123", plan="pro")

Output is JSON to stdout, enriched with level, timestamp, service, location, runtime fields, and Lambda context on invocation.

CloudWatch metrics (EMF)

from wshtlib.metrics import metrics

metrics.count("OrderPlaced")
metrics.put("Duration", 142.5, unit="Milliseconds")
metrics.flush()

metrics is a module-level MetricsContext instance. For isolated contexts (e.g. per-request), instantiate MetricsContext() directly.

Namespace defaults to the METRICS_NAMESPACE env var, falling back to "Wholeshoot".

Request context

from wshtlib import get_context, set_user_id

set_user_id(claims["sub"])
ctx = get_context()  # {"trace_id": ..., "correlation_id": ..., "user_id": ...}

Context is stored in a ContextVar — safe for concurrent async handlers.

FastAPI middleware

from fastapi import FastAPI
from wshtlib.middleware import WshtlibMiddleware

app = FastAPI()
app.add_middleware(WshtlibMiddleware)

Initialises request context, logs method, path, status, duration_ms per request, and injects X-Trace-Id into the response.

Utilities

from wshtlib import require_env, require_https_url, require_secret

db_url = require_env("DATABASE_URL")          # raises RuntimeError if missing/empty
endpoint = require_https_url(require_env("API_URL"))  # raises ValueError if not https
api_key = require_secret("api/key")           # raises RuntimeError if missing/empty, cached

Environment variables

Development

uv sync --group dev
uv run pytest
uv run mypy wshtlib
uv run ruff check wshtlib

License

MIT