Backend Option B: Python API with AWS SAM

What you'll build

The same Notes API as a standalone serverless backend on AWS — API Gateway + Lambda (Python) + DynamoDB in one SAM template — built the way JunctionNet ships production APIs: a clean layered architecture with a middleware factory for dependency injection.

The layered architecture

Every request flows down through the layers and back up. Keeping them separate is what keeps a serverless API testable as it grows:

API Gateway → Lambda
  ▼ handler       parse the request, shape the response (thin)
  ▼ middleware    inject_services — the factory wires dependencies
  ▼ service       business rules (framework-agnostic, easy to test)
  ▼ repository    data access — the only layer that touches the DB
  ▼ model         the domain object
  · schemas       Pydantic: validate in, serialize out

The project structure

Start with the minimum, then let it grow into the shape below. The rule of thumb: one directory per layer, one file that reads the environment, and everything that isn't code — infra, CI, per-stage config — lives at the root where it's easy to review.

notes-api/
├─ template.yaml                       # infra: table, function(s), routes, event bus, IAM
├─ samconfig.toml                      # per-stage deploy config (dev / prod)
├─ requirements.txt                    # runtime deps (boto3 ships with Lambda)
├─ .github/workflows/deploy.yml        # CI/CD: build → sync secrets → sam deploy
├─ tests/                              # unit tests — services with a fake repository
└─ src/
   ├─ service_factory.py               # DI: builds services + repositories (lazily)
   ├─ env_vars.py                      # the ONE place that reads os.environ
   └─ notes/
      ├─ handlers/
      │  ├─ api/notes.py               # HTTP routes — thin
      │  └─ events/on_note_event.py    # EventBridge handler — reacts to events
      └─ application/
         ├─ middlewares.py             # inject_services (the factory middleware)
         ├─ schemas.py                 # Pydantic request/response models
         ├─ services.py                # business logic
         ├─ repositories.py            # data access (the only layer that touches the DB)
         ├─ event_bus.py               # publishes domain events to EventBridge
         └─ models.py                  # domain models

Install the tools

brew install awscli aws-sam-cli    # macOS; see docs for other OSes
aws configure                      # access key + region, e.g. eu-west-1

Declare the infrastructure

# template.yaml
Transform: AWS::Serverless-2016-10-31
Resources:
  NotesTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions: [{ AttributeName: id, AttributeType: S }]
      KeySchema:            [{ AttributeName: id, KeyType: HASH }]
  NotesFn:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: notes.handlers.api.notes.handler
      Runtime: python3.13
      Environment: { Variables: { TABLE: !Ref NotesTable } }
      Policies: [{ DynamoDBCrudPolicy: { TableName: !Ref NotesTable } }]
      Events:
        Api:   { Type: Api, Properties: { Path: /notes, Method: any } }
        ApiId: { Type: Api, Properties: { Path: /notes/{id}, Method: any } }

Build the layers, bottom-up

Schemas — validate in, serialize out

Pydantic models are the contract at the edges. Bad input never reaches your logic:

# src/notes/application/schemas.py
from pydantic import BaseModel, Field

class CreateNote(BaseModel):
    text: str = Field(min_length=1)

class NoteResponse(BaseModel):
    id: str
    text: str

Repository — the only layer that touches data

# src/notes/application/repositories.py
import os, boto3

class NoteRepository:
    def __init__(self, table=None):
        self.table = table or boto3.resource("dynamodb").Table(os.environ["TABLE"])

    def list(self) -> list[dict]:
        return self.table.scan().get("Items", [])

    def create(self, note: dict) -> dict:
        self.table.put_item(Item=note)
        return note

    def delete(self, note_id: str) -> None:
        self.table.delete_item(Key={"id": note_id})

Service — the business rules

Services know nothing about HTTP or AWS — just the domain. That's what makes them trivial to unit-test:

# src/notes/application/services.py
import uuid
from notes.application.schemas import CreateNote, NoteResponse

class NoteService:
    def __init__(self, repository, logger):
        self.repository = repository
        self.logger = logger

    def list_notes(self) -> list[NoteResponse]:
        return [NoteResponse(**n) for n in self.repository.list()]

    def create_note(self, command: CreateNote) -> NoteResponse:
        note = {"id": str(uuid.uuid4()), "text": command.text}
        self.repository.create(note)
        return NoteResponse(**note)

    def delete_note(self, note_id: str) -> None:
        self.repository.delete(note_id)

The middleware factory — dependency injection

Instead of each handler constructing its own services, a factory builds them lazily, and a middleware injects them into every request. Handlers just reach for what they need — and tests can swap in a fake repository.

# src/service_factory.py — builds services + repositories on demand
from aws_lambda_powertools import Logger
from notes.application.repositories import NoteRepository
from notes.application.services import NoteService

class ServiceFactory:
    def __init__(self, logger: Logger):
        self.logger = logger
        self._note_repository = None

    @property
    def note_repository(self) -> NoteRepository:   # lazy: built once, reused
        if self._note_repository is None:
            self._note_repository = NoteRepository()
        return self._note_repository

    def create_note_service(self) -> NoteService:
        return NoteService(repository=self.note_repository, logger=self.logger)

# src/notes/application/middlewares.py — inject deps into the request
from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
from service_factory import ServiceFactory

logger = Logger(service="NotesAPI")

def inject_services(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
    factory = ServiceFactory(logger=logger)
    app.context.setdefault("services", {})
    app.context["services"]["notes"] = factory.create_note_service()
    return next_middleware(app)

Handler — thin routing

The handler wires the middleware chain with app.use(...), validates input with a schema, and delegates to the injected service — no business logic, no direct DB calls:

# src/notes/handlers/api/notes.py
from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, CORSConfig
from aws_lambda_powertools.event_handler.exceptions import BadRequestError
from pydantic import ValidationError
from notes.application.middlewares import inject_services
from notes.application.schemas import CreateNote

app = APIGatewayRestResolver(cors=CORSConfig())
app.use(middlewares=[inject_services])          # + api_error_handler, log_request_response, auth…

@app.get("/notes")
def list_notes():
    return [n.model_dump() for n in app.context["services"]["notes"].list_notes()]

@app.post("/notes")
def create_note():
    try:
        command = CreateNote(**(app.current_event.json_body or {}))
    except ValidationError as e:
        raise BadRequestError(str(e))
    return app.context["services"]["notes"].create_note(command).model_dump(), 201

@app.delete("/notes/<note_id>")
def delete_note(note_id):
    app.context["services"]["notes"].delete_note(note_id)
    return {}, 204

def handler(event, context):
    return app.resolve(event, context)

Test locally, then deploy

sam build
sam local start-api           # http://localhost:3000
sam deploy --guided           # first time; thereafter just: sam deploy

SAM prints your live API Gateway URL. In production, run sam deploy from GitHub Actions on every merge.

The request's User & Organization context

A real API is multi-tenant: every request belongs to a user inside an organization, and one org must never see another's data. You don't parse tokens in the handler — the authorizer validates the credential once at the edge and injects the identity, and the repository scopes every query by org.

Authorization: Bearer <jwt>  /  x-api-key
  ▼ Lambda Authorizer   validate credential, resolve org_id + username (once)
  ▼ requestContext.authorizer.lambda = { org_id, username }
  ▼ middleware          read it into the request context
  ▼ repository          every query filtered by org_id — tenant isolation

Resolve identity once, from the authorizer

# src/notes/application/execution_context.py — who is calling, resolved once
from dataclasses import dataclass

@dataclass(frozen=True)
class RequestContext:
    org_id: str
    username: str

def from_event(event) -> RequestContext:
    claims = event["requestContext"]["authorizer"]["lambda"]
    return RequestContext(org_id=claims["org_id"], username=claims["username"])

The middleware reads it off the event and puts it on the request alongside the services, so handlers just reach for it:

# middlewares.py — inject identity + services
def inject_services(app, next_middleware):
    ctx = from_event(app.current_event.raw_event)      # from the authorizer, never the body
    app.context["ctx"] = ctx
    factory = ServiceFactory(logger=logger, context=ctx)
    app.context.setdefault("services", {})["notes"] = factory.create_note_service()
    return next_middleware(app)

And the repository — the only layer that touches data — filters by org_id on every call:

# repositories.py — every query scoped by org
from boto3.dynamodb.conditions import Key

def list(self, org_id: str) -> list[dict]:
    return self.table.query(
        KeyConditionExpression=Key("org_id").eq(org_id)).get("Items", [])

The database connection

DynamoDB is the easy case: there's no connection to manage — every call is an IAM-signed HTTPS request, so the repository just holds a boto3 client and Lambda scaling is a non-issue. The moment you want relational data (Postgres on RDS/Aurora, or Supabase) two things change: you open a real connection, and you have to think about how many.

Reuse the connection across warm invocations

Lambda freezes the container between requests and thaws it for the next one. Build the engine once at module scope (or behind a cache) so warm invocations reuse it instead of reconnecting on every call — and pull the credentials from Secrets Manager, never from code:

# src/notes/application/repositories.py — relational variant (SQLAlchemy)
import os, json, functools, boto3
from sqlalchemy import create_engine, text

@functools.lru_cache(maxsize=1)          # built once per warm container, then reused
def _engine():
    secret = boto3.client("secretsmanager").get_secret_value(
        SecretId=os.environ["DB_SECRET_ARN"])
    url = json.loads(secret["SecretString"])["connection_string"]
    return create_engine(url, pool_pre_ping=True, pool_size=1, max_overflow=0)

class NoteRepository:
    def list(self) -> list[dict]:
        with _engine().connect() as conn:
            rows = conn.execute(text("SELECT id, text FROM notes"))
            return [dict(r._mapping) for r in rows]

Secrets & API tokens: GitHub → SSM / Secrets Manager → SAM

Never commit a token and never paste one into template.yaml. The clean pattern has one source of truth in AWS and a pipeline that wires it in at deploy time. The flow:

GitHub Secrets / Variables
  │  # CI job, on merge to main
  ▼
AWS SSM Parameter Store  ·  AWS Secrets Manager   # source of truth at deploy + runtime
  │  # {{resolve:ssm:…}} / {{resolve:secretsmanager:…}}  or  --parameter-overrides
  ▼
SAM template.yaml  →  Lambda env vars (injected at deploy)
  │
  ▼  # …or fetched at cold start, for secrets that rotate
running Lambda

The pipeline: push the value, then deploy

GitHub holds only what CI needs to authenticate. App secrets can be kept in AWS directly, or seeded from GitHub Secrets in an idempotent step so the repo stays the single place a value is set:

# .github/workflows/deploy.yml (excerpt)
- name: Configure AWS credentials
  uses: aws-actions/configure-aws-credentials@v4
  with:
    role-to-assume: ${{ secrets.AWS_DEPLOY_ROLE_ARN }}   # OIDC — no long-lived keys
    aws-region: eu-west-1

- name: Sync app secret → SSM / Secrets Manager
  run: |
    aws ssm put-parameter --name "/notes/$STAGE/anthropic_api_key" \
      --value "$ANTHROPIC_API_KEY" --type SecureString --overwrite
  env:
    STAGE: prod
    ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}   # from GitHub Secrets

- name: Deploy
  run: sam deploy --config-env "$STAGE" --parameter-overrides Stage="$STAGE"

SAM injects it — config baked in, secrets by reference

# template.yaml
Parameters:
  Stage: { Type: String, Default: dev }
Resources:
  NotesFn:
    Type: AWS::Serverless::Function
    Properties:
      Environment:
        Variables:
          STAGE: !Ref Stage
          TABLE: !Ref NotesTable
          # non-secret config, resolved from SSM at deploy time:
          API_BASE_URL:  !Sub '{{resolve:ssm:/notes/${Stage}/api_base_url}}'
          # a secret's ARN — the VALUE is fetched at cold start (survives rotation):
          DB_SECRET_ARN: !Sub '{{resolve:ssm:/notes/${Stage}/db_secret_arn}}'
      Policies:                                        # least-privilege, scoped to this stage
        - SSMParameterReadPolicy: { ParameterName: !Sub 'notes/${Stage}/*' }
        - AWSSecretsManagerGetSecretValuePolicy:
            SecretId: !Sub '{{resolve:ssm:/notes/${Stage}/db_secret_arn}}'

The event bus: services that react to each other

Some work shouldn't happen inline with the request — indexing a note for search, sending a notification, metering usage. Instead of the service calling those directly (and waiting on them), it publishes a domain event to EventBridge and moves on. Consumers subscribe by pattern and run in their own Lambda — the producer never knows they exist.

Publish from the service

# src/notes/application/event_bus.py — a thin publisher
import os, json, boto3

class EventBus:
    def __init__(self, client=None):
        self.client = client or boto3.client("events")
        self.bus = os.environ["EVENT_BUS"]

    def publish(self, detail_type: str, detail: dict) -> None:
        self.client.put_events(Entries=[{
            "EventBusName": self.bus,
            "Source": "notes.api",          # {app}.{service} — dotted, lowercase
            "DetailType": detail_type,      # "NoteCreated" — PascalCase, past tense
            "Detail": json.dumps(detail),
        }])

The factory hands the bus to the service (one more property), and the service fires the event after the write succeeds — fire-and-forget, off the request path:

# in services.py
def create_note(self, command: CreateNote) -> NoteResponse:
    note = {"id": str(uuid.uuid4()), "text": command.text}
    self.repository.create(note)
    self.event_bus.publish("NoteCreated", {"id": note["id"]})
    return NoteResponse(**note)

React in a separate handler

# src/notes/handlers/events/on_note_event.py
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.data_classes import event_source, EventBridgeEvent

logger = Logger(service="NotesEvents")

@event_source(data_class=EventBridgeEvent)
def handler(event: EventBridgeEvent, context):
    if event.detail_type == "NoteCreated":
        logger.info("note created", extra={"id": event.detail["id"]})
        # …update a search index, notify, meter usage — the slow work goes here

Wire it in the template

The bus, the producer's permission to publish, and the consumer's subscription rule — SAM creates the rule and the invoke permission for you:

# template.yaml
Resources:
  NotesBus:
    Type: AWS::Events::EventBus
    Properties: { Name: !Sub 'notes-${Stage}' }

  NotesFn:                                   # the producer
    Type: AWS::Serverless::Function
    Properties:
      Environment: { Variables: { EVENT_BUS: !Ref NotesBus } }
      Policies:
        - EventBridgePutEventsPolicy: { EventBusName: !Ref NotesBus }

  OnNoteEventFn:                             # the consumer
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: notes.handlers.events.on_note_event.handler
      Runtime: python3.13
      Events:
        NoteEvents:
          Type: EventBridgeRule
          Properties:
            EventBusName: !Ref NotesBus
            Pattern:
              source: [ notes.api ]
              detail-type: [ NoteCreated, NoteDeleted ]

A CLAUDE.md for the SAM backend

# CLAUDE.md

## What this is
Serverless API on AWS: API Gateway + Lambda (Python 3.13) + DynamoDB, via SAM.
Layered DDD: handlers → services → repositories → models; Pydantic schemas at the edges.

## Structure  # one directory per layer; grow by adding a folder, not fattening a file
src/
  service_factory.py                      # DI: builds services + repositories (lazy)
  env_vars.py                             # the one place that reads os.environ
  notes/
    handlers/{api,events}/                # thin entry points (API Gateway, EventBridge)
    application/{services,schemas}/        # business logic + Pydantic validation
    infrastructure/{models,repositories}/  # ORM + org_id-scoped data access
    domain/                               # value types, state machines, enums
template.yaml  ·  samconfig.toml  ·  tests/

## Commands
- Build / Local / Deploy: sam build / sam local start-api / sam deploy --config-env <stage>
- Logs: sam logs -n NotesFn --tail

## Architecture rules
- Handlers are thin: parse, delegate to an injected service, shape the response. No logic, no DB.
- Dependencies come from the ServiceFactory via the inject_services middleware —
  handlers never construct services or repositories directly.
- Only repositories touch the database. Only schemas (Pydantic) cross the edges.
- All infra in template.yaml; IAM = least privilege (never "Action": "*").

## Auth & multi-tenancy
- org_id + username come from the authorizer (requestContext), never the request body.
- Scope every query by org_id in the repository — no route may leak another tenant's data.

## Secrets & config
- Never commit secrets or paste them in template.yaml. Source of truth is AWS:
  SSM Parameter Store (config) / Secrets Manager (secrets). CI syncs GitHub → AWS.
- Config: resolve via {{resolve:ssm:/notes/${Stage}/...}} into env at deploy.
- Rotating secrets: put the ARN in env, fetch the value at cold start (not resolve).

## Database
- Relational: build the engine once at module scope; small pool; go through a
  pooler (RDS Proxy / Supavisor). Run migrations in CI before the deploy.

## Events (EventBridge)
- source = {platform}.{service}.{module} (dotted, stable); detail-type = {Entity}{PastTenseVerb}.
- Publish from the SERVICE, after the write commits. Off the request path. One event per fact.
- Consumers: narrow pattern; idempotent (key on event id) + DLQ. Bursty work goes
  event → SQS → Lambda (batch + retries). Mirror the bus to OpenSearch for analytics.

## Verify
sam local start-api and curl the routes; check CloudWatch logs after deploy.

Adding the next service — same shape

Every service is the same DDD skeleton, so adding one is copy-the-shape, not invent-a-new-one. A new orders service mirrors the existing layout exactly:

services/orders/                 # a new SAM stack — mirrors documents/
├─ src/
│  ├─ handlers/{api,event_bus}/   # thin entry points
│  ├─ application/{services,schemas}/
│  ├─ infrastructure/{models,repositories}/   # org_id-scoped
│  ├─ domain/
│  ├─ factories.py                # lazy DI
│  └─ env_vars.py
├─ template.yaml                  # its own stack + least-privilege IAM
└─ samconfig.toml

• Copy the skeleton, don't redesign it — handlers → services → repositories → models, factory DI, Pydantic at the edges. Consistency is the feature.
• Read shared infra by reference — {{resolve:ssm:/platform/*}} for the bus, buckets, VPC (Part 2); never hardcode ids.
• Keep the conventions — org_id-scoped repositories, EventBridge source/detail-type naming, least-privilege IAM in its own template.

Ask for a feature the way an engineer would

With the layers in place, don't just say "add filtering." State the contract, route each piece to its layer, lean on Pydantic for validation, keep IAM least-privilege, and end with a way to verify:

> Add "list notes by tag with pagination", following the layers:
  Contract: GET /notes?tag=work&limit=20&cursor=<token>
  → { "items": [...], "next_cursor": "..." | null }.
  schemas/ : a ListNotesQuery (Pydantic) — tag optional str,
  limit int 1–100 default 20, cursor optional str.
  repositories/ : list_notes(tag, limit, cursor) does the DynamoDB
  query + pagination — a Query on a GSI, never a Scan.
  services/ : returns a ListNotesResponse; the handler stays thin
  and uses the injected service.
  Update template.yaml only if a new IAM action is needed, and keep
  it least-privilege. Don't touch the existing routes.
  Show me the diff and a `sam local` curl for each case first.

• Contract first — the exact query params and JSON response shape, including the pagination cursor.
• One instruction per layer — schema, repository, service, handler — so logic lands where it belongs and the factory keeps wiring it.
• Validate with Pydantic — bounds like limit 1–100 belong in the schema, not scattered in the handler.
• Name the AWS gotcha — "Query on a GSI, never a Scan" and "least-privilege IAM" steer it away from the expensive, insecure defaults.
• Verify before deploy — ask for the diff and a sam local curl so you test locally, not in production.

Supercharge Claude Code with the AWS Agent Toolkit

Install the AWS Agent Toolkit: it bundles the AWS MCP Server (Claude Code can query AWS docs, discover skills, and run real AWS calls) plus curated AWS skills:

# in Claude Code
/plugin marketplace add aws/agent-toolkit-for-aws
/plugin install aws-core@agent-toolkit-for-aws
/reload-plugins

> Add a GET /notes/{id} route following the existing layers: a
  repository.get(id), a service.get_note(id) returning NoteResponse,
  and a thin handler that uses the injected service. Keep the IAM
  policy least-privilege. Show me the diff before deploying.