How a domain Lambda is packaged and started under the AWS Lambda Web Adapter, and the test fixtures. Back to the README.
There is no Lambda handler and no Mangum. The AWS Lambda Web Adapter is an external
extension that starts before the application, turns each invoke into an ordinary HTTP
request against 127.0.0.1:$AWS_LWA_PORT, and turns the response back. The application is a
normal ASGI server, so the identical image runs on Lambda, in a local container, and
anywhere else.
# app/posts/entrypoint.py
from webbpulse.lambda_entry import run_uvicorn
from webbpulse.logging import configure_logging
from webbpulse.otel import configure_tracing
from app.posts import build_app
def main() -> None:
configure_logging(level="INFO", service="posts", environment="staging")
configure_tracing("webbpulse-staging-posts", environment="staging")
run_uvicorn(build_app())
if __name__ == "__main__":
main()run_uvicorn binds AWS_LWA_PORT, falling back to PORT and then 8080, which is the
adapter's own precedence. Binding a port the adapter is not polling is the most common Web
Adapter misconfiguration and it presents as the readiness check never passing and the
function timing out with no application logs at all.
The Dockerfile is one COPY from a pinned public image. Version 1.0.1 is current, and the
image is multi-arch, so the same line serves arm64 and x86_64:
# syntax=docker/dockerfile:1.7
FROM public.ecr.aws/docker/library/python:3.13-slim AS build
WORKDIR /build
COPY requirements.txt .
RUN --mount=type=secret,id=codeartifact_token \
PIP_INDEX_URL="https://aws:$(cat /run/secrets/codeartifact_token)@webbpulse-432410731887.d.codeartifact.us-west-2.amazonaws.com/pypi/python/simple/" \
pip install --no-cache-dir --target /deps -r requirements.txt
FROM public.ecr.aws/docker/library/python:3.13-slim
COPY --from=public.ecr.aws/awsguru/aws-lambda-adapter:1.0.1 /lambda-adapter /opt/extensions/lambda-adapter
COPY --from=build /deps /var/task
COPY app /var/task/app
ENV PYTHONPATH=/var/task \
PYTHONUNBUFFERED=1 \
AWS_LWA_PORT=8080 \
AWS_LWA_READINESS_CHECK_PATH=/health \
AWS_LWA_ASYNC_INIT=true
WORKDIR /var/task
CMD ["python", "-m", "app.posts.entrypoint"]Each line there is a real failure mode:
/opt/extensions/lambda-adapteris the required destination. Lambda only starts binaries it finds in/opt/extensions. Copied anywhere else the adapter never runs, the function has no handler, and every invoke times out.- The CodeArtifact token is a BuildKit secret mount, never a build arg or
ENV. Both of those persist into the image and are visible indocker history. PYTHONUNBUFFERED=1keeps log lines from sitting in a buffer while the execution environment is frozen between invokes and arriving attributed to a later request.AWS_LWA_ASYNC_INIT=truelets a slow import finish inside Lambda's 10 second init window instead of counting against the first invoke.AWS_LWA_READINESS_CHECK_PATH=/healthmust point at a route that does no I/O. The adapter's default is/.
Pytest fixtures for a moto-backed table and a TestClient. Enable them from a service's
conftest.py:
pytest_plugins = ["webbpulse.testing"]| Fixture or helper | What it gives you |
|---|---|
aws_credentials |
Placeholder credentials and region, so a mis-scoped mock cannot reach a real account |
dynamodb_resource |
A moto-mocked DynamoDB resource, with the package's cached resource cleared on both sides |
dynamodb_reset_hooks |
Override it to have dynamodb_resource clear a product's own memoised resource too |
create_table(...) |
One on-demand table, with an optional range key, TTL, GSIs, stream or whole request |
rate_limit_table |
The rate-limits table shaped exactly as Terraform creates it |
test_client(app, source_ip=...) |
A TestClient whose requests carry a realistic API Gateway request context |
make_request_context_headers(...) |
That header on its own, in either payload shape |
rsa_key |
A module-scoped 2048-bit RSA key, so a suite generates one rather than one per test |
fake_kms |
A FakeKms holding rsa_key, ready to drive a KmsSigner |
FakeKms is the KMS stand-in the identity tests run on, exported for a service's own. It
answers get_public_key and sign in the shapes the real client returns and signs for
real, so a token minted through it verifies against its JWK. It holds either one key, which
answers for any key id, or a mapping of key id to key for a rotation test. It records
get_public_key_calls and sign_calls, takes a failing set of key ids that raise the way
a deleted key does, reports a key_spec, and der_for(key_id) returns exactly the bytes
kid_for_der hashes. MessageType="DIGEST" is honoured: the message is signed as the
digest it already is, never hashed again.
create_table shapes the common case from keyword arguments, and takes raw CreateTable
pieces for everything else: attribute_definitions for the attributes an index keys on,
global_secondary_indexes, stream_specification, and request for a whole keyword
mapping. Keys in request win over the ones the helper builds, TableName is always the
name argument, and the waiting and the TTL still happen, so a product holding its own
specs stops hand-rolling both:
create_table(
dynamodb_resource,
"projects",
attribute_definitions=[{"AttributeName": "owner", "AttributeType": "S"}],
global_secondary_indexes=[
{
"IndexName": "by-owner",
"KeySchema": [{"AttributeName": "owner", "KeyType": "HASH"}],
"Projection": {"ProjectionType": "ALL"},
}
],
stream_specification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"},
)
create_table(dynamodb_resource, spec.table_name(prefix), request=spec.create_table_request(prefix))TTL is never part of CreateTable, so ttl_attribute stays a keyword argument and applies
alongside a request mapping.
dynamodb_resource clears the package's cached resource on both sides of the mock. A
product that memoises its own boto3 resource has one more cache to drop, and rather than
wrapping the fixture it overrides dynamodb_reset_hooks, which yields the callables the
fixture runs on setup and teardown:
@pytest.fixture
def dynamodb_reset_hooks() -> list[Callable[[], None]]:
from app.common.db.dynamo.client import reset_clients
return [reset_clients]The package's own reset always runs first and is not in the list. Anything depending on
dynamodb_resource, rate_limit_table included, picks the override up.
test_client is the one worth knowing about. Without the injected context header a
TestClient request has no API Gateway context at all, so client_ip falls back to the
peer address and a service's rate limit tests pass while never covering the branch that
actually runs in production.