LambdaApi — Lambdas behind API Gateway

sosw.lambda_api.LambdaApi is a Processor subclass for AWS Lambda functions that serve HTTP requests behind Amazon API Gateway. One Lambda backs a whole (micro)API: you declare a route table in the config, implement the handlers as methods, and LambdaApi takes care of event parsing, authorization, routing, response rendering and the error contract.

Both API Gateway payload formats are supported: REST API (v1.0) and HTTP API (v2.0) proxy events.

Minimal example

import json

from sosw.app import LambdaGlobals, get_lambda_handler
from sosw.components.exceptions import NotFoundError
from sosw.lambda_api import LambdaApi


class Processor(LambdaApi):

    DEFAULT_CONFIG = {
        'auth_enabled': True,
        'routes':       {
            'GET /things':            'list_things',
            'GET /things/{thing_id}': 'get_thing',
            'POST /things':           'create_thing',
        },
        'things_dynamo_db_config': {
            'table_name': 'things',
            'row_mapper': {'thing_id': 'S'},
        },
    }


    def list_things(self, event, **kwargs):
        return {'things': self.get_ddbc('things').get_by_scan()}


    def get_thing(self, event, thing_id=None, **kwargs):
        things = self.get_ddbc('things').get_by_query({'thing_id': thing_id})
        if not things:
            raise NotFoundError(f"Thing {thing_id} does not exist")
        return {'thing': things[0]}


    def create_thing(self, event, **kwargs):
        thing = json.loads(event.get('body') or '{}')
        self.get_ddbc('things').put(thing)
        return self.make_response({'created': thing['thing_id']}, status_code=201)


global_vars = LambdaGlobals()
lambda_handler = get_lambda_handler(Processor, global_vars)

The regular sosw.app.get_lambda_handler() is used — LambdaApi inherits the full warm start behavior, and the compiled router is cached on the instance for the lifetime of the container.

Routing

The routes config parameter maps 'METHOD /path' keys to names of handler methods:

• Paths use the API Gateway resource syntax: {param} matches one path segment, the greedy {proxy+} matches the rest of the path including slashes.

• Handlers receive the raw event as the first positional argument and the extracted path parameters (URL-decoded) as keyword arguments: def get_thing(self, event, thing_id=None, **kwargs).

• Routes without placeholders are matched before parametrized ones, so GET /things/count wins over GET /things/{thing_id} regardless of declaration order.

• Request paths are normalized before matching: the query string and the trailing slash are stripped (/things/ and /things?page=2 both match /things).

• If the deployed API serves routes under a stage or mount prefix, list it in path_prefixes (e.g. 'path_prefixes': ['/prod'] turns /prod/things into /things).

A handler may return:

• any JSON-serializable object — rendered as a 200 response;

• a pre-rendered dict containing statusCode (typically built with make_response(body, status_code=..., headers=...)) — passed through to API Gateway untouched;

• or raise an ApiError subclass — rendered as the corresponding HTTP error envelope.

Authorization

LambdaApi expects the API Gateway authorizer (Cognito User Pool authorizer on REST APIs, JWT authorizer on HTTP APIs) to verify the token signature. The class then works with the claims the authorizer injected into the event:

• REST API v1: requestContext.authorizer.claims;

• HTTP API v2: requestContext.authorizer.jwt.claims.

With auth_enabled: True (the secure default) every request must carry claims:

1. Missing claims → 401.

2. Expired token → 401. The exp claim is tolerated in every shape the authorizers deliver: numeric epoch, numeric string (HTTP API v2), and the legacy custom-authorizer string format ('Wed Aug 06 09:02:12 UTC 2025'). A missing or unparsable exp is not treated as expired — the authorizer remains the authority on token validity.

3. When the authorized_groups config parameter is non-empty, the caller must belong to at least one of the listed Cognito groups (the cognito:groups claim, tolerated as a real list, the stringified '[admins users]' v2 form, or a comma/space-separated string) → otherwise 403.

Validated claims are available to handlers as self.claims (reset on every invocation). With auth_enabled: False claims are still extracted best-effort into self.claims, but nothing is enforced.

Error responses never include claim contents — the reasons are logged server-side only.

For per-route rules, override the check_route_access(route, claims) hook — it runs after route matching and before the handler:

class Processor(LambdaApi):

    WRITE_METHODS = ('POST', 'PUT', 'PATCH', 'DELETE')


    def check_route_access(self, route, claims):
        if route.startswith(self.WRITE_METHODS) and 'admins' not in self.parse_groups(
                claims.get('cognito:groups')):
            raise ForbiddenError("Write access requires the admins group")

Responses, CORS and JSON encoding

make_response(body, status_code=200, headers=None) renders the API Gateway proxy response {'statusCode', 'headers', 'body'}. Headers are the config-driven cors_headers (the defaults allow every origin — tighten them for production) merged with the per-call headers.

The JSON encoder handles the types DynamoDB-backed code returns without ceremony:

• Decimal → int when integral, else float;

• datetime / date → ISO-8601 string;

• set / frozenset → sorted list.

Override the static method json_default to support more types (e.g. UUID).

The error contract

Handlers (and the pipeline itself) communicate failures by raising the ApiError hierarchy from sosw.components.exceptions:

Exception	HTTP status	error.code	Raised by the pipeline when
BadRequestError	400	BAD_REQUEST	the event is not an API Gateway proxy event
UnauthorizedError	401	UNAUTHORIZED	claims are missing or the token expired
ForbiddenError	403	FORBIDDEN	the caller is in none of authorized_groups
NotFoundError	404	NOT_FOUND	no configured route matches method + path
ConflictError	409	CONFLICT	(your handlers)
ServerError	500	SERVER_ERROR	any unexpected exception

Every error renders the same machine-readable envelope:

{
    "error": {
        "code":    "NOT_FOUND",
        "message": "No route configured for: GET /nope"
    }
}

A custom message travels in message; raise ConflictError(f"Thing {thing_id} already exists") produces HTTP 409 with the envelope carrying your text.

The exact semantics of the pipeline:

• Authorization runs before routing. A request with missing/invalid claims receives 401 even for a path that does not exist — unauthenticated callers cannot probe the route table.

• Unknown route or method → 404. There is no automatic 405: a known path with an unconfigured method is just as much “not found”.

• Only ApiError maps to specific statuses. Any other exception escaping a handler is logged server-side with the full traceback, counted in stats, and rendered as the generic 500 envelope — internals never leak to the client.

Note

Unlike regular sosw Processors, which must fail fast and loud, LambdaApi.__call__ intentionally converts exceptions into well-formed HTTP responses: an API Gateway integration must always answer the client. This is the API contract layer, not silent error handling — everything is logged and counted server-side.

REST API (v1) vs HTTP API (v2)

LambdaApi absorbs the differences between the two payload formats; for reference:

Aspect	REST API (payload v1.0)	HTTP API (payload v2.0)
Method / path	httpMethod + path	requestContext.http.method + rawPath
Claims	requestContext.authorizer.claims	requestContext.authorizer.jwt.claims
exp claim type	numeric	numeric string
cognito:groups claim	list	stringified '[admins users]'
Fallback	—	routeKey ('GET /things') for trimmed events

Statistics

LambdaApi counts per-container statistics through the standard self.stats mechanism:

• api_calls — every invocation;

• api_route_get_things_thing_id — per matched route (snake_case of the route key, e.g. 'GET /things/{thing_id}');

• api_errors_4xx / api_errors_5xx — per error family.

Deployment

See the HTTP API with LambdaApi tutorial for a complete SAM template with a Cognito JWT authorizer and curl-level verification of the whole contract.

API reference

View Licence Agreement

sosw - a framework for bootstrapping AWS Lambda functions

The MIT License (MIT)
Copyright (C) 2026  sosw core contributors <info@sosw.app>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

class sosw.lambda_api.LambdaApi(*args, **kwargs)

Base Processor for AWS Lambda functions behind API Gateway. Supports both the REST API (payload version 1.0) and the HTTP API (payload version 2.0) proxy event shapes.

Provides a declarative router configured with the routes config parameter, Cognito JWT claims extraction and validation, config-driven CORS headers, JSON response rendering with a DynamoDB-friendly encoder, and a uniform machine-readable error envelope.

Subclasses implement the handler methods and declare them in routes:

from sosw.app import LambdaGlobals, get_lambda_handler
from sosw.lambda_api import LambdaApi

class Processor(LambdaApi):

    DEFAULT_CONFIG = {
        'auth_enabled': True,
        'routes':       {
            'GET /things':            'list_things',
            'GET /things/{thing_id}': 'get_thing',
        },
    }

    def list_things(self, event, **kwargs):
        return {'things': self.get_ddbc('things').scan()}

    def get_thing(self, event, thing_id=None, **kwargs):
        return {'thing': self.get_ddbc('things').get_by_query({'thing_id': thing_id})}

global_vars = LambdaGlobals()
lambda_handler = get_lambda_handler(Processor, global_vars)

Handlers receive the raw event as the first positional argument and the path parameters extracted from the request path (per the {param} placeholders of the matched route) as keyword arguments. A handler may return:

• any JSON-serializable object - rendered as a 200 response body;

• a pre-rendered response dict containing statusCode (e.g. built with make_response() using a custom status code or extra headers) - passed through to API Gateway untouched;

• or raise sosw.components.exceptions.ApiError (or a subclass) - rendered as the corresponding HTTP error envelope.

Note

Unlike regular sosw Processors which are required to fail fast and loud, __call__ of this class intentionally converts exceptions to well-formed HTTP error responses: an API Gateway integration must always return a valid response to the client. This is the API contract layer, not silent error handling: every exception is logged server-side (unexpected ones with the full traceback), counted in self.stats, and unexpected exceptions are never exposed to the client beyond a generic 500 envelope.

Configuration parameters of this class:

• routes - dict mapping 'METHOD /path' route keys to names of handler methods. Paths support API Gateway resource syntax placeholders: {param} and greedy {proxy+}.

• auth_enabled - when True (secure default), requests must carry Cognito claims injected by the API Gateway authorizer, otherwise 401 is returned.

• authorized_groups - optional list of Cognito group names. When non-empty, the caller must be a member of at least one of them (the cognito:groups claim), otherwise 403.

• cors_headers - dict of headers attached to every rendered response.

• path_prefixes - optional list of prefixes (e.g. stage or mount prefixes) stripped from the request path before route matching.

parse_event(event: dict) → tuple

Extract the HTTP method and the normalized request path from an API Gateway proxy event.

Supported event shapes:

• REST API v1: httpMethod + path;

• HTTP API v2: requestContext.http.method + rawPath;

• trimmed v2 events carrying only routeKey ('GET /things') as the last resort.

The path is normalized (query string and trailing slash removed) and the first matching entry of the path_prefixes config parameter is stripped from it.

Parameters:

event (dict) – Raw Lambda event.

Return type:

tuple

Returns:

Tuple of (method, path), method upper-cased.

Raises:

BadRequestError – If the event does not look like an API Gateway proxy event.

get_router() → list

Return the compiled router, building it on the first request of the container.

The router is built from the routes config parameter by build_router() and cached on the instance for the container lifetime, so warm invocations skip parsing and regex compilation entirely.

Return type:

list

Returns:

List of compiled route entries.

build_router() → list

Compile the routes config parameter into a list of route entries.

Each 'METHOD /path': 'handler_name' config pair becomes a dict with the original route key, the upper-cased method, the normalized path template, the compiled path pattern and the bound handler method. Routes without path parameters are ordered before parametrized ones, so GET /things/count wins over GET /things/{thing_id} regardless of the declaration order. Within each group the config declaration order is preserved.

Return type:

list

Returns:

List of compiled route entries.

Raises:

ValueError – If a route key is malformed or a handler method is missing.

static compile_path_pattern(path_template: str) → Pattern

Compile a path template with API Gateway resource syntax into a regex pattern.

{param} placeholders match a single path segment as the named group param. The greedy {proxy+} form matches the rest of the path (including slashes) as proxy.

Parameters:

path_template (str) – Path template, e.g. '/things/{thing_id}'.

Return type:

re.Pattern

Returns:

Compiled pattern matching the whole request path.

match_route(method: str, path: str) → tuple

Find the route entry matching the request and extract its path parameters.

Path parameter values are URL-decoded. Requests that match no configured route (either by path or by method) raise sosw.components.exceptions.NotFoundError -> 404.

Parameters:

• method (str) – Upper-cased HTTP method of the request.

• path (str) – Normalized request path.

Return type:

tuple

Returns:

Tuple of (route entry, dict of extracted path parameters).

Raises:

NotFoundError – When no configured route matches the request.

static get_claims(event: dict) → dict

Extract Cognito JWT claims from either API Gateway authorizer event shape.

• REST API v1 (Cognito User Pool authorizer): requestContext.authorizer.claims

• HTTP API v2 (JWT authorizer): requestContext.authorizer.jwt.claims

Both the id and access token claim sets are supported - the method does not require any specific claim to be present.

Parameters:

event (dict) – API Gateway proxy event.

Return type:

dict

Returns:

The claims dictionary, or an empty dict if the event carries none.

validate_authorization(event: dict) → dict

Validate the caller identity from the claims injected by the API Gateway authorizer.

The authorizer has already verified the JWT signature; this method enforces that claims are present, the token is not expired, and (when the authorized_groups config parameter is non-empty) the caller belongs to at least one authorized Cognito group.

Error responses never include claim contents; the reasons are logged server-side only.

Parameters:

event (dict) – API Gateway proxy event.

Return type:

dict

Returns:

The validated claims.

Raises:

• UnauthorizedError – When claims are missing or the token is expired -> 401.

• ForbiddenError – When the caller is not in any authorized group -> 403.

static token_expired(exp: Any) → bool

Check whether the exp claim points to the past.

Tolerated formats: numeric epoch (the standard JWT claim), numeric strings (the form HTTP API v2 authorizers deliver), and the legacy custom-authorizer string format 'Wed Aug 06 09:02:12 UTC 2025'. A missing or unparsable claim is not treated as expired - the API Gateway authorizer remains the authority on token validity.

Parameters:

exp – Value of the exp claim in any supported format.

Return type:

bool

static parse_groups(raw_groups: Any) → list

Parse the cognito:groups claim into a list of group names.

Tolerated formats: a real list (REST API v1), the stringified '[admins users]' form delivered by HTTP API v2 JWT authorizers, and comma- or space-separated plain strings.

Parameters:

raw_groups – Raw value of the cognito:groups claim.

Return type:

list

check_route_access(route: str, claims: dict) → None

Per-route access control hook, called after route matching and before the handler.

The base implementation allows everything. Override in subclasses to implement custom RBAC (e.g. permission tables in DynamoDB, interface grants, organization scopes) and raise sosw.components.exceptions.ForbiddenError (-> 403) or sosw.components.exceptions.UnauthorizedError (-> 401) to deny the request.

Parameters:

• route (str) – The matched route key exactly as declared in the routes config, e.g. 'GET /things/{thing_id}'.

• claims (dict) – Validated claims of the caller (empty dict when auth_enabled is off and the event carries no claims).

make_response(body: Any = None, status_code: int = 200, headers: dict = None) → dict

Render an API Gateway proxy response with the configured CORS headers.

The body is JSON-serialized with json_default(), which supports the types DynamoDB and datetime-heavy code commonly return (Decimal, datetime, date, set).

Parameters:

• body – JSON-serializable payload, or None for an empty body.

• status_code (int) – HTTP status code of the response.

• headers (dict) – Extra headers merged over the configured cors_headers.

Return type:

dict

Returns:

API Gateway proxy response: {'statusCode', 'headers', 'body'}.

make_error_response(error: ApiError) → dict

Render the uniform error envelope for an ApiError and count the error statistics.

The envelope is {'error': {'code': <MACHINE_CODE>, 'message': <human readable>}}. Statistics: api_errors_4xx / api_errors_5xx per status code family.

Parameters:

error (ApiError) – The error to render.

Return type:

dict

Returns:

API Gateway proxy response with the error envelope.

static json_default(value: Any) → Any

JSON encoder fallback for types that json.dumps does not handle natively.

• Decimal (as returned by DynamoDB) - int when integral, else float;

• datetime / date - ISO-8601 string;

• set / frozenset - sorted list (plain list when elements are not sortable).

Override in subclasses to support more types (e.g. UUID).

Parameters:

value – Value that JSON could not serialize.

Returns:

A JSON-serializable representation.

Raises:

TypeError – For unsupported types (standard json contract).

strip_path_prefix(path: str) → str

Strip the first matching entry of the path_prefixes config parameter from the path.

Useful when the deployed API serves routes under a stage or mount prefix (e.g. /prod/things -> /things with 'path_prefixes': ['/prod']).

Parameters:

path (str) – Normalized request path.

Return type:

str

static route_stats_key(route: str) → str

Build the snake_case self.stats counter name for a route key.

E.g. 'GET /things/{thing_id}' -> 'api_route_get_things_thing_id'.

Parameters:

route (str) – Route key as declared in the routes config.

Return type:

str