Capstone Project

Build, test, and deploy a Task Management API with FastAPI.

Introduction

Welcome to the capstone project. Across this course you learned REST design, request and response modelling, validation, error handling, OpenAPI documentation, testing, and deployment. Now you will put all of it into a single service: a Task Management API built with FastAPI, organised into models, controllers, and routers, covered by unittest, and deployed to AWS Lambda behind API Gateway with the CDK. Data lives in an in-memory store on purpose, so you can run and deploy the whole thing without provisioning a database.

Project Purpose

The goal is a small API you could defend in a code review. Not a toy with everything in one file, and not an over-engineered framework either. Every task carries a title, an optional description, a lifecycle status and a priority. The API exposes full CRUD plus filtering and pagination on the collection.

What You Are Practising

This project exercises the parts of the course that matter most in real work:

  • Resource modelling, Designing URLs, verbs, and status codes that follow HTTP semantics
  • Schema validation, Using Pydantic to reject bad input at the edge instead of deep in the code
  • Separation of concerns, Keeping transport, business logic, and data shapes in different modules
  • Partial updates, Implementing PATCH correctly, so unspecified fields are untouched
  • Testing, Driving the app through a test client and asserting on real HTTP responses
  • Deployment, Shipping the same app to Lambda without changing a line of application code

The endpoints you will build are the ones almost every service has. Get them right here, and the pattern transfers directly to orders, invoices, users, or whatever domain you work in next.

The API Surface

Design the contract before writing code. Six endpoints, mounted under /api:

MethodPathSuccessPurpose
GET/api/health200Liveness probe, returns a plain OK
GET/api/tasks200List tasks, filter by status, paginate
POST/api/tasks201Create a task, returns the stored resource
GET/api/tasks/{task_id}200Fetch one task, 404 when absent
PATCH/api/tasks/{task_id}200Partial update, only supplied fields change
DELETE/api/tasks/{task_id}204Delete a task, empty body on success

Why PATCH and not PUT

PUT replaces a resource in full, so a client that omits description is asking you to clear it. PATCH applies a partial change. Since clients here usually flip a single field, such as moving a task to in_progress, PATCH is the honest verb. Lesson 4 covers this trade-off in depth.

First Steps

Step 1: Create Your Project Folder

Start with a dedicated folder for the capstone:

$ mkdir task-management-api
$ cd task-management-api

Step 2: Create the Virtual Environment

Set up an isolated environment. The code uses StrEnum and the str | None union syntax, so Python 3.11 or newer is required:

$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install --upgrade pip

Step 3: Install the Dependencies

Create a requirements.txt for what the service needs at runtime:

fastapi
starlette
environs
core-mixins
uvicorn
click

And a requirements-dev.txt for what only the tests need:

# Test runner with branch coverage, exposed through manager.py
core-tests

# Required by starlette.testclient; tests only, never imported by the service.
httpx
$ pip install -r requirements.txt
$ pip install -r requirements-dev.txt
Recommendation

Keep test-only packages out of the runtime file from day one. httpx is needed by the Starlette test client, never by the service itself. When you later package this for Lambda, that single split is the difference between shipping your dependencies and shipping your test harness too.

Project Structure

The layout separates the three concerns you will keep separate in every serious API: the shapes of the data, the logic that manipulates it, and the HTTP transport that exposes it.

task-management-api/
├── manager.py                      # CLI entry point
├── requirements.txt
├── requirements-dev.txt
├── service/
│   ├── __init__.py
│   └── api/
│       ├── __init__.py             # Shared logger configuration
│       ├── main.py                 # create_app() application factory
│       ├── models/
│       │   ├── __init__.py
│       │   └── task.py             # Pydantic schemas
│       ├── controllers/
│       │   ├── __init__.py         # Controller registry
│       │   └── task.py             # TaskController, in-memory store
│       └── routers/
│           ├── __init__.py         # Router registration
│           ├── health.py
│           └── task.py             # CRUD endpoints
└── tests/
    └── unit/
        └── api/
            ├── base.py             # Shared TestClient setup
            ├── tests_health.py
            └── tests_tasks.py

models/

Pydantic schemas describing what goes in and what comes out. No logic, no storage, just shapes and their validation rules.

controllers/

The business logic. Knows how tasks are stored, filtered, and updated. Knows nothing about HTTP, which is exactly why it is easy to test.

routers/

The HTTP layer. Parses query parameters, calls a controller, turns a missing result into a 404. Thin by design.

Where a Request Goes

ClientRouterControllerModelsPOST /api/tasksvalidate TaskCreate422 on bad inputcreate(payload)assign id, timestampstask dictserialize TaskResponse201 Created

Figure 1: Validation happens before your logic runs, serialization after it returns

Implementing models/task.py

Start with the data shapes. Three models, because creating, updating, and returning a task are three different contracts. Collapsing them into one model is the most common mistake in FastAPI codebases.

models/task.py - Complete Implementation

# -*- coding: utf-8 -*-

"""Pydantic models describing task payloads and responses."""

from datetime import datetime
from enum import StrEnum

from pydantic import BaseModel
from pydantic import ConfigDict
from pydantic import Field


class TaskStatus(StrEnum):
    """Lifecycle state of a task."""

    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"


class TaskPriority(StrEnum):
    """Relative importance of a task."""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class TaskCreate(BaseModel):
    """Payload to create a task."""

    title: str = Field(..., min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=2000)
    status: TaskStatus = TaskStatus.TODO
    priority: TaskPriority = TaskPriority.MEDIUM


class TaskUpdate(BaseModel):
    """Payload to partially update a task (only provided fields change)."""

    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=2000)
    status: TaskStatus | None = None
    priority: TaskPriority | None = None


class TaskResponse(BaseModel):
    """Task representation returned to clients."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    description: str | None
    status: TaskStatus
    priority: TaskPriority
    created_at: datetime
    updated_at: datetime

What's Happening Here?

  • StrEnum for wire values, TaskStatus subclasses StrEnum, so its members compare equal to plain strings and serialize as "todo" rather than "TaskStatus.TODO". Member names are uppercase by convention, values stay lowercase because that is what travels over the wire.
  • Three models, three contracts, TaskCreate requires a title and defaults the rest.TaskUpdate makes every field optional so a client can send just one. TaskResponse adds the server-owned fields, id, created_at, and updated_at, which a client must never set.
  • Validation at the boundary, Field(..., min_length=1, max_length=200) means an empty or oversized title never reaches your logic. FastAPI turns the failure into a structured 422 for free.
  • ConfigDict, not class Config, model_config = ConfigDict(from_attributes=True) is the Pydantic v2 spelling. The inner class Config you may see in older tutorials is v1 and is deprecated.

Implementing controllers/task.py

The controller owns the data and the rules. Notice what it does not import: no FastAPI, no HTTP status codes, no request objects. That independence is what lets you swap the in-memory dictionary for a real database later without touching a single endpoint.

controllers/task.py - Complete Implementation

# -*- coding: utf-8 -*-

"""Task controller backing the CRUD endpoints with an in-memory store."""

from datetime import datetime
from datetime import timezone

from service.api.models.task import TaskCreate
from service.api.models.task import TaskStatus
from service.api.models.task import TaskUpdate


class TaskController:
    """Stores tasks in memory and assigns their incremental identifiers."""

    def __init__(self) -> None:
        self._items: dict[int, dict] = {}
        self._next_id = 1

    def clear(self) -> None:
        """Drop every task and reset the identifier sequence."""

        self._items.clear()
        self._next_id = 1

    def list_tasks(
        self, status: TaskStatus | None, limit: int, offset: int
    ) -> list[dict]:
        """Return tasks ordered by id, optionally filtered by status.

        :param status: Keep only tasks in this status when given.
        :param limit: Maximum number of tasks to return.
        :param offset: Number of tasks to skip before collecting results.
        """

        items = list(self._items.values())
        if status is not None:
            items = [task for task in items if task["status"] == status]

        items.sort(key=lambda task: task["id"])
        return items[offset : offset + limit]

    def create(self, data: TaskCreate) -> dict:
        """Store a new task and return it with its generated id and timestamps."""

        now = datetime.now(timezone.utc)

        task = {
            "id": self._next_id,
            "created_at": now,
            "updated_at": now,
            **data.model_dump(),
        }

        self._items[self._next_id] = task
        self._next_id += 1
        return task

    def get(self, task_id: int) -> dict | None:
        """Return the task with this id, or None when it does not exist."""

        return self._items.get(task_id)

    def update(self, task_id: int, data: TaskUpdate) -> dict | None:
        """Apply the provided fields to a task, or return None when absent."""

        task = self._items.get(task_id)
        if task is None:
            return None

        task.update(data.model_dump(exclude_unset=True))
        task["updated_at"] = datetime.now(timezone.utc)
        return task

    def delete(self, task_id: int) -> bool:
        """Remove a task, reporting whether one was actually deleted."""

        return self._items.pop(task_id, None) is not None

What's Happening Here?

  • None means not found, get() and update() return None when the task does not exist, and delete() returns a boolean. The controller reports facts, the router decides that the fact deserves a 404. That is the separation working.
  • exclude_unset is the heart of PATCH, data.model_dump(exclude_unset=True) returns only the fields the client actually sent. Without it, every unset optional would come back as None and wipe the stored values. This single argument is the difference between a correct and a destructive PATCH.
  • Timezone-aware timestamps, datetime.now(timezone.utc) produces an aware datetime.datetime.utcnow() returns a naive one and is deprecated since Python 3.12, so avoid it in new code.
  • clear() exists for the tests, Resetting the store and the id counter between tests keeps each case independent. You will see it used in setUp() shortly.

controllers/__init__.py - The Registry

Endpoints need a single shared controller instance. A small registry gives you one, created lazily on first use:

# -*- coding: utf-8 -*-

"""Controllers package; holds the in-memory business logic."""

from service.api.controllers.task import TaskController


class Controller:
    """Registry of the controllers, each created on first access."""

    _task_ctrl: TaskController | None = None

    @classmethod
    def task_controller(cls) -> TaskController:
        """
        Return the task controller, instantiating it on first use.
        :returns: TaskController object.
        """

        if cls._task_ctrl is None:
            cls._task_ctrl = TaskController()
        return cls._task_ctrl

Why a Registry Instead of a Module-Level Instance

Creating the controller at import time makes it a hidden global that is awkward to reset or replace. A classmethod accessor keeps construction lazy, gives the tests one obvious place to reach for, and leaves room to grow: when you add a second resource, it becomes another method on the same registry.

Implementing routers/task.py

Now the HTTP layer. Every handler is a few lines: take validated input, call the controller, translate the result into a response or an error. If a handler grows past that, the logic belongs in the controller.

routers/task.py - Complete Implementation

# -*- coding: utf-8 -*-

"""
Task Management router that exposes CRUD endpoints for tasks. Data is
kept in a simple in-memory store.
"""

from __future__ import annotations

from typing import List
from typing import Optional

from fastapi import APIRouter
from fastapi import HTTPException
from fastapi import Query
from starlette.status import HTTP_201_CREATED
from starlette.status import HTTP_204_NO_CONTENT
from starlette.status import HTTP_404_NOT_FOUND

from service.api.controllers import Controller
from service.api.models.task import TaskCreate
from service.api.models.task import TaskResponse
from service.api.models.task import TaskStatus
from service.api.models.task import TaskUpdate

tasks_router = APIRouter(prefix="/tasks", tags=["Tasks"])


@tasks_router.get(
    path="",
    response_model=List[TaskResponse],
    description="List tasks.")
async def list_tasks(
    status: Optional[TaskStatus] = Query(default=None, description="Filter by status"),
    limit: int = Query(default=20, ge=1, le=100, description="Max items to return"),
    offset: int = Query(default=0, ge=0, description="Items to skip"),
):
    """Return a page of tasks, optionally narrowed to a single status."""

    return Controller.task_controller().list_tasks(
        status=status,
        limit=limit,
        offset=offset,
    )


@tasks_router.post(
    path="",
    response_model=TaskResponse,
    status_code=HTTP_201_CREATED,
    description="Create a task.",
)
async def create_task(payload: TaskCreate):
    """Create a task from the payload and return it as stored."""

    return Controller.task_controller().create(payload)


@tasks_router.get(
    path="/{task_id}",
    response_model=TaskResponse,
    description="Get a task by id.")
async def get_task(task_id: int):
    """Return a single task, raising 404 when it does not exist."""

    task = Controller.task_controller().get(task_id)
    if task is None:
        raise HTTPException(
            status_code=HTTP_404_NOT_FOUND,
            detail=f"Task {task_id} not found",
        )

    return task


@tasks_router.patch(
    path="/{task_id}",
    response_model=TaskResponse,
    description="Partially update a task.")
async def update_task(task_id: int, payload: TaskUpdate):
    """Apply the supplied fields to a task, raising 404 when it does not exist."""

    task = Controller.task_controller().update(task_id, payload)
    if task is None:
        raise HTTPException(
            status_code=HTTP_404_NOT_FOUND,
            detail=f"Task {task_id} not found",
        )

    return task


@tasks_router.delete(
    path="/{task_id}",
    status_code=HTTP_204_NO_CONTENT,
    description="Delete a task.")
async def delete_task(task_id: int):
    """Delete a task, raising 404 when it does not exist."""

    if not Controller.task_controller().delete(task_id):
        raise HTTPException(
            status_code=HTTP_404_NOT_FOUND,
            detail=f"Task {task_id} not found",
        )

    return None

What's Happening Here?

  • Prefix and tags, APIRouter(prefix="/tasks", tags=["Tasks"]) keeps the path in one place and groups these operations under a single heading in the generated OpenAPI docs.
  • response_model does two jobs, It documents the schema in OpenAPI and filters the outgoing payload, so a stray internal key in the controller dictionary can never leak to a client.
  • Query constraints are documentation, Query(default=20, ge=1, le=100) rejects alimit of 0 or 10000 before your code runs, and publishes those bounds in the schema. Clients see the contract instead of guessing.
  • Correct status codes, 201 for a created resource, 204 with no body for a delete,404 when an id does not resolve. Lesson 1 and Lesson 7 argued for these, this is where you honour them.
  • Enum-typed query parameters, Typing status as Optional[TaskStatus] means FastAPI validates the value against the enum and renders a dropdown in Swagger UI. An unknown status gets a422, not an empty list.

Wiring It Together

Two small files connect everything. The routers package collects every router with the prefix it should be mounted under, and the application factory builds the app and attaches them.

routers/__init__.py

# -*- coding: utf-8 -*-

"""Routers package; collects the routers exposed by the service."""

import os
from typing import List, Tuple

from fastapi import APIRouter

from .health import health_router
from .task import tasks_router

BASE_PATH = os.environ.get("BASE_PATH", "")
API_BASE_PATH = f"{BASE_PATH}/api"

routers: List[Tuple[str, APIRouter]] = [
    (API_BASE_PATH, health_router),
    (API_BASE_PATH, tasks_router),
]

api/__init__.py - The Shared Logger

Configure logging once, in the package that everything else lives under, so every module inherits it.core-mixins provides a get_logger helper that applies a sensible format and level:

# -*- coding: utf-8 -*-

"""API package; configures the logger shared by the service modules."""

import logging
import os

from core_mixins.logger import get_logger

logger = get_logger(
    reset_handlers=True,
    log_level=int(os.getenv("LOGGER_LEVEL", str(logging.INFO))),
    propagate=False,
)

reset_handlers=True clears handlers another library may have attached, which is what stops the duplicate log lines you often see under uvicorn. propagate=False keeps records from bubbling to the root logger and being emitted twice. Reading the level from LOGGER_LEVEL means you can raise verbosity in a deployed environment by changing a variable, with no redeploy of code.

main.py - The Application Factory

# -*- coding: utf-8 -*-

"""Application factory for the Task Management API."""

import os
from contextlib import suppress

from environs import Env
from fastapi import FastAPI

from .routers import routers


def create_app(api_title: str, debug: bool = False, **kwargs):
    """Build a FastAPI app with the environment loaded and routers attached."""

    with suppress(OSError):
        Env().read_env(os.path.join(os.getcwd(), ".env"))

    app_ = FastAPI(title=api_title, debug=debug, **kwargs)
    add_routers(app_)
    return app_


def add_routers(app_: FastAPI):
    """Register every configured router under its base path."""

    for prefix, router in routers:
        app_.include_router(router, prefix=prefix)

What's Happening Here?

  • A factory, not a global app, create_app() builds a fresh application on each call. Tests get their own instance with their own title, and nothing is constructed as a side effect of importing a module.
  • Registration is data, Routers live in a list of (prefix, router) tuples. Adding a resource later means appending one line, not editing the factory.
  • Configurable mount point, BASE_PATH lets you serve the API under a different root without touching any route, which matters when something upstream adds a path segment.
  • Optional .env, suppress(OSError) means a missing .env is not an error. Locally you get file-based configuration, in the cloud the platform supplies the variables.
  • One CLI, several sources, CommandCollection(sources=[cli_main, cli_tests]) merges your own commands with the runner shipped by core-tests, so run-api-service, run-tests, andrun-coverage all hang off the same manager.py.

manager.py - The CLI Entry Point

A small click CLI at the project root starts the service with uvicorn, and folds in the test commands from core-tests so everything runs through one entry point:

# -*- coding: utf-8 -*-

import logging
import os

from click import group
from click.core import CommandCollection
from core_tests.tests.runner import cli_tests
from environs import Env
from uvicorn import run


@group()
def cli_main():
    pass


@cli_main.command("run-api-service")
def run_api_service():
    from service.api.main import create_app

    run(
        app=create_app(
            api_title=os.environ.get("API_NAME", "Service API"),
            debug=Env().bool("DEBUG", False)
        ),
        log_level=int(os.getenv("LOGGER_LEVEL", logging.INFO)),
        host=os.environ.get("HOST", "0.0.0.0"),
        port=os.environ.get("PORT", 8080)
    )


if __name__ == "__main__":
    Env().read_env(os.path.join(os.getcwd(), ".env"))
    CommandCollection(sources=[cli_main, cli_tests])()
Recommendation

Parse booleans with Env().bool("DEBUG", False) rather than bool(os.environ.get("DEBUG")). The second form treats the string "false" as True, because any non-empty string is truthy in Python. It is a classic bug that only bites in production, where someone finally sets the variable.

Running and Exercising the API

Start the Service

$ python manager.py run-api-service

The service listens on 0.0.0.0:8080. Open http://localhost:8080/docs for the Swagger UI that FastAPI generates from your models and route declarations, no extra work required. Lesson 8 explains what it builds and how to shape it.

Create a Task

Creating a task returns 201 Created with the stored resource, including the fields the server owns:

curl --location --request POST 'http://0.0.0.0:8080/api/tasks' \
--header 'Content-Type: application/json' \
--data '{
    "title": "Write the documentation",
    "description": "Cover every task endpoint",
    "status": "todo",
    "priority": "high"
}'
Expected Output:
{
  "id": 1,
  "title": "Write the documentation",
  "description": "Cover every task endpoint",
  "status": "todo",
  "priority": "high",
  "created_at": "2026-07-21T14:32:10.481293Z",
  "updated_at": "2026-07-21T14:32:10.481293Z"
}

List, Filter, Paginate, Update, Delete

# List everything (limit defaults to 20)
curl 'http://0.0.0.0:8080/api/tasks'

# Second page of two
curl 'http://0.0.0.0:8080/api/tasks?limit=2&offset=2'

# Only completed tasks
curl 'http://0.0.0.0:8080/api/tasks?status=done'

# Move one task forward, leaving every other field untouched
curl --location --request PATCH 'http://0.0.0.0:8080/api/tasks/1' \
--header 'Content-Type: application/json' \
--data '{"status": "in_progress"}'

# Remove it (204, empty body)
curl --location --request DELETE 'http://0.0.0.0:8080/api/tasks/1'

Watch Validation Work

Send an empty title and you get a 422 that names the offending field, generated entirely from the model:

{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "title"],
      "msg": "String should have at least 1 character",
      "input": "",
      "ctx": {"min_length": 1}
    }
  ]
}

You Wrote Zero Validation Code

Every rule in that response came from the Field declarations in models/task.py. This is why the models are worth designing carefully: they are simultaneously your validation, your documentation, and your serialization contract.

Testing the API

Test through the HTTP surface, not around it. Starlette's TestClient drives the real application in-process, so your assertions cover routing, validation, status codes, and serialization all at once, with no server to start.

tests/unit/api/base.py - Shared Setup

# -*- coding: utf-8 -*-

from unittest import TestCase

from starlette.testclient import TestClient

from service.api.main import create_app


class ServiceApiBaseTestCases(TestCase):
    """ Base test case for the API Service tests """

    @classmethod
    def setUpClass(cls) -> None:
        super(ServiceApiBaseTestCases, cls).setUpClass()

        app = create_app(api_title="ApiService-TestsExecution")
        cls.client = TestClient(app)
        cls.app = app

tests/unit/api/tests_tasks.py - A Representative Selection

# -*- coding: utf-8 -*-

from service.api.controllers import Controller

from .base import ServiceApiBaseTestCases


class TaskRouterTestCases(ServiceApiBaseTestCases):
    """Tests for the Task Management router."""

    def setUp(self) -> None:
        super().setUp()
        Controller.task_controller().clear()

    def test_create_and_get_task(self):
        response = self.client.post("/api/tasks", json={"title": "Write docs"})
        self.assertEqual(response.status_code, 201)

        body = response.json()
        self.assertEqual(body["title"], "Write docs")
        self.assertEqual(body["status"], "todo")
        self.assertEqual(body["priority"], "medium")
        self.assertIsNone(body["description"])

        response = self.client.get(f"/api/tasks/{body['id']}")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["title"], "Write docs")

    def test_get_missing_task_returns_404(self):
        response = self.client.get("/api/tasks/999")
        self.assertEqual(response.status_code, 404)

    def test_list_pagination(self):
        for index in range(5):
            self.client.post("/api/tasks", json={"title": f"Task {index}"})

        response = self.client.get("/api/tasks", params={"limit": 2, "offset": 2})
        data = response.json()

        self.assertEqual(len(data), 2)
        self.assertEqual([task["title"] for task in data], ["Task 2", "Task 3"])

    def test_patch_updates_only_provided_fields(self):
        created = self.client.post(
            "/api/tasks",
            json={"title": "A", "description": "keep me"}
        ).json()

        response = self.client.patch(
            f"/api/tasks/{created['id']}",
            json={"status": "in_progress"}
        )

        self.assertEqual(response.status_code, 200)
        body = response.json()

        self.assertEqual(body["status"], "in_progress")
        self.assertEqual(body["title"], "A")
        self.assertEqual(body["description"], "keep me")

    def test_validation_error_on_empty_title(self):
        response = self.client.post("/api/tasks", json={"title": ""})
        self.assertEqual(response.status_code, 422)

What's Happening Here?

  • Isolation through clear(), setUp() resets the store before each test. Without it, tests would pass or fail depending on the order they run in, which is the fastest way to lose trust in a suite.
  • Assert on the contract, The tests check status codes and response bodies, the things a client actually depends on. They never reach into _items. That means you can rewrite the controller entirely and the tests still tell you whether the API still works.
  • Defaults are behaviour worth testing, test_create_and_get_task asserts that an omitted status becomes "todo" and priority becomes "medium". Defaults are part of your public contract.
  • Test the failures too, A suite that only covers the happy path tells you nothing about the day something goes wrong. Cover the 404s and the 422, they are the responses clients hit in production.

Run the suite through the same CLI, then run it again under branch coverage. run-coverage prints a console summary and writes an HTML report you can open in a browser:

$ python manager.py run-tests
$ python manager.py run-coverage

Your Target

Aim for every endpoint and every failure branch covered: create, get, list, filter, paginate, patch, delete, all three 404s, and the 422. That is roughly ten tests, and it is achievable here precisely because the code is layered. Lesson 14 goes further into coverage, load testing, and monitoring.

Deploying to AWS

The final step: put the API on the internet without rewriting it. Mangum adapts your ASGI application to the Lambda event format, so the same create_app() that serves uvicorn locally serves API Gateway in the cloud.

Deployed Architecture

USRClientsAGWAPI GatewayREST, proxyλLambdaFastAPI + MangumCWCloudWatchLogsHTTPSinvokelogsresponse

Figure 2: One function behind a proxy REST API, with no database to provision

handler.py - The Lambda Entry Point

# -*- coding: utf-8 -*-

"""AWS Lambda entry point that serves the FastAPI app through API Gateway."""

import os

from environs import Env
from mangum import Mangum

from service.api.main import create_app

env = Env()

app = create_app(
    api_title=os.environ.get("API_NAME", "Task Management API"),
    debug=env.bool("DEBUG", False),
)

handler = Mangum(app, lifespan="off")

That is the entire adapter. Mangum translates API Gateway events into ASGI calls and back.lifespan="off" is correct here because this app registers no startup or shutdown handlers; if you add any later, turn it back on.

infrastructure/stack.py - The CDK Stack

# -*- coding: utf-8 -*-

import logging
import pathlib

from aws_cdk import CfnOutput
from aws_cdk import Duration
from aws_cdk import Stack
from aws_cdk import aws_apigateway as apigw
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as _lambda
from core_aws_cdk.stacks.lambdas.assets import ZipAssetCode


class TaskManagementBackendStack(Stack):
    """Backend stack: the API Lambda and the REST API in front of it."""

    def create_resources(self) -> None:
        """Create every resource that makes up the backend."""

        # Execution role for the FastAPI Lambda: basic execution and CloudWatch Logs.
        role = iam.Role(
            self,
            id="TaskManagement-LambdaRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    "service-role/AWSLambdaBasicExecutionRole"
                ),
            ],
        )

        # Single Lambda running the FastAPI app via Mangum.
        api_fcn = _lambda.Function(
            self, "TaskManagement-ApiFn",
            function_name="task-management-api",
            description="API Service for Task Management",
            runtime=_lambda.Runtime.PYTHON_3_12,
            handler="handler.handler",
            role=role,
            timeout=Duration.seconds(29),
            memory_size=512,
            environment={
                "API_NAME": "Task Management API",
                "LOGGER_LEVEL": str(logging.INFO),
            },
            code=ZipAssetCode(
                project_directory=pathlib.Path.cwd(),
                work_dir=pathlib.Path.cwd() / "infrastructure" / "lambdas" / "api",
                includes=["handler.py"],
                include_project_folders=["service"],
            ),
        )

        api = apigw.LambdaRestApi(
            self, "TaskManagement-Backend",
            rest_api_name="task-management-api",
            description="Task Management REST API",
            handler=api_fcn,
            proxy=True,
            endpoint_types=[apigw.EndpointType.REGIONAL],
            default_cors_preflight_options=apigw.CorsOptions(
                allow_origins=apigw.Cors.ALL_ORIGINS,
                allow_methods=apigw.Cors.ALL_METHODS,
                allow_headers=["Authorization", "Content-Type"],
            ),
            deploy_options=apigw.StageOptions(
                throttling_burst_limit=100,
                throttling_rate_limit=10,
            ),
        )

        CfnOutput(
            self, "TaskManagement-BackendUrl",
            value=api.url,
            export_name="TaskManagement-BackendUrl",
        )

ZipAssetCode comes from core-aws-cdk and handles packaging at synth time: it installs therequirements.txt found in work_dir, copies in the files listed by includes plus the project folders named in include_project_folders, and zips the result. That is why the Lambda can import service.api.main exactly as your local process does. Add it to the CDK requirements:

aws-cdk-lib
core-aws-cdk

The Lambda gets its own requirements.txt under infrastructure/lambdas/api/, holding only what runs in the cloud:

fastapi
starlette
environs
core-mixins
mangum
Recommendation

Notice what that file leaves out: uvicorn, click, and httpx. Nothing in the Lambda path imports them, since API Gateway replaces the server and Mangum replaces the CLI. Also note that the packaging step runs pip with python3.12 to match the function runtime. Packages with compiled extensions, pydantic-core among them, ship per-platform wheels, so a package built against the wrong interpreter fails at import time inside Lambda rather than at deploy time.

app.py - The CDK Entry Point

# -*- coding: utf-8 -*-

import os

import aws_cdk as cdk
from environs import Env

from infrastructure.stack import TaskManagementBackendStack

Env().read_env(os.path.join(os.getcwd(), ".env"))
app = cdk.App()


stack = TaskManagementBackendStack(
    scope=app,
    id="TaskManagement-Backend",
    env=cdk.Environment(
        account=os.getenv("CDK_DEFAULT_ACCOUNT"),
        region=os.getenv("CDK_DEFAULT_REGION"),
    ),
)

stack.create_resources()
app.synth()

What's Happening Here?

  • 29 seconds, not 30, The Lambda timeout matches the hard integration limit of REST API Gateway. Setting the function higher achieves nothing, the gateway gives up first and returns 504.
  • Proxy integration, proxy=True forwards every path and method to the one function, so FastAPI keeps doing the routing. You define routes in one place, not twice.
  • Throttling at the edge, throttling_rate_limit=10 with a burst of 100 caps traffic before it reaches your code. Cheaper and safer than rate limiting inside the application, as discussed in Lesson 9.
  • Least privilege, The execution role carries only AWSLambdaBasicExecutionRole, which grants CloudWatch Logs and nothing else. Grant more only when a resource actually needs it.
  • CORS is infrastructure here, Preflight responses are configured on the gateway rather than in FastAPI. Tighten allow_origins to your real domains before anything goes public.

Set CDK_DEFAULT_ACCOUNT and CDK_DEFAULT_REGION, bootstrap once per account and region, then deploy:

$ cdk bootstrap
$ cdk synth
$ cdk diff
$ cdk deploy

The stack outputs TaskManagement-BackendUrl. Your endpoints live under the stage path:

curl 'https://<api-id>.execute-api.<region>.amazonaws.com/prod/api/health'
curl 'https://<api-id>.execute-api.<region>.amazonaws.com/prod/api/tasks'

Important: What the In-Memory Store Means on Lambda

Locally, tasks live as long as the process. On Lambda the consequences are sharper: each execution environment holds its own copy of the controller. Two concurrent requests may hit two different environments, so aGET can return 404 for a task a POST just created, and the id counter restarts on every cold start.

This is not a bug to work around, it is the lesson. Stateless compute demands external state. The moment you need the deployed API to behave, you reach for DynamoDB or a relational database and move the controller's dictionary behind that. Lesson 20 covers exactly that step.

Extending the Project

You now have a working, tested, deployed API. Each extension below maps to a lesson in this course, and each one fits the existing structure without restructuring what you built.

Persistence

Replace the dictionary in TaskController with a real store. Because nothing else knows how tasks are saved, the routers and models stay untouched. See Lesson 20.

Authentication

Add JWT bearer tokens as a FastAPI dependency, then scope tasks to their owner. See Lesson 5 and Lesson 6.

RFC 7807 Errors

Swap the plain detail string for a structured problem document with a type, title, and instance. See Lesson 7.

Cursor Pagination

Offset pagination drifts when items are inserted mid-scan. Move to cursors and return aLink header. See Lesson 23.

Webhooks

Emit a signed task.completed event when a task reaches done, with HMAC signatures and retries. See Lesson 10.

Versioning

Mount the router under /api/v1 and plan how a breaking change to TaskResponse would ship. See Lesson 13.

Key Takeaways

  • Layers earn their keep - Models validate, controllers decide, routers translate. Each one is replaceable because none of them reaches into the others.
  • Separate models per operation - Create, update, and response are different contracts. One shared model forces you to make required fields optional and leak server-owned fields.
  • exclude_unset makes PATCH correct - It is the difference between changing one field and silently erasing the rest.
  • HTTP semantics are not decoration - 201 with the resource, 204 with no body,404 for a missing id, 422 for bad input. Clients build retry and error handling on these.
  • Test through the client - Asserting on status codes and bodies lets you rewrite internals freely while keeping a real guarantee about the contract.
  • Deployment should change nothing - An ASGI app plus an adapter runs on a server or on Lambda. If deploying forces you to edit application code, the boundary was drawn in the wrong place.
  • Stateless compute needs external state - The in-memory store is fine for learning and immediately insufficient in production. Knowing exactly why is the point of building it this way.
Where to Go From Here
  • Make it yours - Swap tasks for a domain you care about. The structure carries over unchanged.
  • Add one extension properly - Pick a single item from the list above and finish it, tests included. One well-built feature says more in an interview than six half-built ones.
  • Put it in your portfolio - A layered, tested, deployed API with a clear README is a genuinely strong artifact to show.

Bonus: The Complete Reference Implementation

Every file in this lesson comes from a working repository you can clone, run, and deploy. Use it to check your work, or as the starting point for the extensions above.

The repository includes:

  • The full layered service - Models, controllers, and routers exactly as built here
  • The complete test suite - Every endpoint and failure branch covered, run with manager.py
  • CDK infrastructure - The Lambda, REST API, and packaging setup ready to deploy
  • A documented README - Setup, every endpoint with curl examples, and the deployment steps
View the repository:Task Management API on GitLab
Web APIsLesson Capstone