Build a Learning Platform API

Build an async FastAPI backend with PostgreSQL, Redis caching, and layered architecture.

Introduction

30 Second Summary

Every app you use daily relies on a backend API to fetch, store, and update data. The speed and reliability of that API determines whether the experience feels instant or sluggish.

In this project, you will build a backend API for a learning platform using FastAPI, async PostgreSQL, and Redis caching. You'll structure the code with a production-style layered architecture and test it all through interactive API docs.

What You'll Build

A fully functional async API that manages courses and lessons, with a layered architecture, database migrations, and intelligent caching.

By the end of this project, you'll have:

A FastAPI application with versioned API routes for managing courses and lessons.
Async PostgreSQL database access using SQLAlchemy 2.0 with Alembic migrations.
A Redis caching layer implementing the cache-aside pattern with TTL-based expiration and write-through invalidation.
Secret Mission: Add cursor-based pagination and a Redis-cached leaderboard endpoint showing the most popular courses by lesson count.

Are there any prerequisites?

You'll need Python 3.11+ and Docker Desktop installed on your machine. Some experience with Python and REST APIs is helpful.

Before We Start

Before We Start...

Set Up the Project and Infrastructure

Every backend API needs solid infrastructure before you write the first line of application code. Your learning platform API will use PostgreSQL for persistent data storage and Redis for caching.

In this step, you'll scaffold the project directories following a layered architecture pattern. You'll also spin up database and cache containers with Docker Compose and install all the Python packages your API needs.

In this step, get ready to:

Set up the project directory structure and Docker services.
Create a Python virtual environment and install dependencies.
Configure application settings with environment variables.

Create the project directory and start Docker services

A layered architecture organizes your code into distinct folders, each with a specific job. Your API will have routers, services, and repositories as separate layers.

What is layered architecture?

Routers handle incoming HTTP requests and return responses. Services contain your business logic (like caching decisions). Repositories talk directly to the database.

Each layer only talks to the one below it. This separation makes your code easier to test, change, and reason about.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder, all the layered architecture subdirectories, and Python package files by running these commands:

mkdir learn-platform-api
cd learn-platform-api
mkdir -p app/core app/db app/models app/repositories app/services app/schemas app/api/v1
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/repositories/__init__.py app/services/__init__.py app/schemas/__init__.py app/api/__init__.py app/api/v1/__init__.py

What do these commands do?

mkdir -p creates nested directories in one go. Each folder maps to a layer: app/api/ for routers, app/services/ for business logic, app/repositories/ for database access.
touch creates empty __init__.py files. Python needs these files to treat each folder as an importable package.

Confirm the directories were created by running:

ls app/

You should see __init__.py, api, core, db, models, repositories, schemas, and services in the output.

Using Windows PowerShell?

The mkdir -p and touch commands are for macOS/Linux terminals and Git Bash. If you're using PowerShell, open Git Bash instead (it comes with Git for Windows).

Help me create these directories on Windows.

Now that the project structure is ready, you need the database and cache services running locally. Docker Compose lets you define and start both services from a single YAML file.

Create a file called docker-compose.yml in the root of your learn-platform-api folder and paste in this configuration:

# Define the two services your API depends on
services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: learnapi
      POSTGRES_PASSWORD: learnapi
      POSTGRES_DB: learnplatform
    ports:
      - "5432:5432"
    volumes:
      # Persist database data across container restarts
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - "6379:6379"

# Named volume for PostgreSQL data persistence
volumes:
  postgres_data:

What does this configuration do?

The db service runs PostgreSQL 16 with a user, password, and database pre-configured. Port 5432 is mapped so your Python app can connect from the host machine.
The redis service runs Redis 7 on port 6379. No extra configuration needed for local development.
The postgres_data volume keeps your database data alive even if you stop and restart the container.

Save docker-compose.yml.
Start both containers by running this command:

docker compose up -d

The -d flag runs the containers in the background so you can keep using your terminal.

Verify both containers are running by running:

docker compose ps

You should see two containers listed with a status of running. One for db (PostgreSQL) and one for redis (Redis).

Containers not starting?

Make sure Docker Desktop is running. You should see the Docker icon in your system tray or menu bar.
If port 5432 or 6379 is already in use, another service is occupying it. Stop that service or change the port mapping in docker-compose.yml.
Check the logs with docker compose logs to see detailed error messages.

Help me troubleshoot my Docker Compose containers.

✔️ Awesome, I've got everything!

Great. Make sure you've saved docker-compose.yml and both containers show running in docker compose ps.

ⓧ I'd like to double check the full code

services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: learnapi
      POSTGRES_PASSWORD: learnapi
      POSTGRES_DB: learnplatform
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - "6379:6379"

volumes:
  postgres_data:

Create a virtual environment and install dependencies

A virtual environment keeps this project's Python packages separate from your system Python. This prevents version conflicts between different projects.

Create a virtual environment inside your learn-platform-api folder by running:

python3 -m venv venv

Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On macOS, make sure you used source before the path. Running venv/bin/activate without source won't work.
On Windows, if you see a permissions error, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell first, then try activating again.

Help me activate my Python virtual environment.

Install all the project dependencies by running this command:

pip install fastapi==0.136.1 uvicorn==0.46.0 "sqlalchemy[asyncio]==2.0.49" asyncpg==0.31.0 alembic==1.18.4 redis==7.4.0 pydantic==2.13.3 pydantic-settings==2.9.1

What are all these packages?

FastAPI is the web framework that handles HTTP requests and serves your API.
Uvicorn is the ASGI server that runs your FastAPI app.
SQLAlchemy (with the async extra) is the ORM that maps Python classes to database tables. asyncpg is the async PostgreSQL driver it uses under the hood.
Alembic handles database migrations so you can evolve your schema over time.
redis is the Python client for talking to your Redis cache.
Pydantic validates request and response data. pydantic-settings loads configuration from environment variables.

You should see Successfully installed followed by a list of all installed packages at the end of the output.

Seeing installation errors?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt.
If asyncpg fails to install, you may need a C compiler. On macOS, run xcode-select --install. On Windows, install the Visual C++ Build Tools.
Confirm you are using Python 3.11 or later by running python3 --version.

Help me fix my pip install errors.

Create the configuration module

Your API needs to know where to find the database and cache. Instead of hardcoding connection strings throughout your code, you'll use a central configuration file that loads values from environment variables.

Create a new file at app/core/config.py and paste in this code:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Connection string for the async PostgreSQL driver
    database_url: str = "postgresql+asyncpg://learnapi:learnapi@localhost:5432/learnplatform"
    # Redis connection for caching
    redis_url: str = "redis://localhost:6379/0"
    app_name: str = "LearnPlatformAPI"
    debug: bool = False

    # Tell Pydantic to load values from a .env file
    model_config = {"env_file": ".env"}


# Create a single Settings instance the whole app will import
settings = Settings()

What does this code do?

Settings extends Pydantic's BaseSettings, which automatically reads environment variables and maps them to class fields. The field database_url maps to the DATABASE_URL environment variable.
model_config = {"env_file": ".env"} tells Pydantic to look for a .env file and load variables from it. This is the Pydantic v2 way to configure settings.
The settings instance at the bottom is created once and imported everywhere. Every module that needs config just writes from app.core.config import settings.

Save app/core/config.py.

The config file references a .env file that doesn't exist yet. This file stores the actual values for your local environment.

Create a file called .env in the root of your learn-platform-api folder and paste in these values:

DATABASE_URL=postgresql+asyncpg://learnapi:learnapi@localhost:5432/learnplatform
REDIS_URL=redis://localhost:6379/0
APP_NAME=LearnPlatformAPI
DEBUG=true

What do these values mean?

DATABASE_URL points to the PostgreSQL container you just started. The asyncpg part tells SQLAlchemy to use the async driver.
REDIS_URL connects to your local Redis container on port 6379, using database 0 (the default).
DEBUG=true enables verbose SQL logging, which helps you see exactly what queries your app sends to the database.

Save .env.
Verify that Pydantic can load your settings by running:

python3 -c "from app.core.config import settings; print(settings.app_name)"

You should see LearnPlatformAPI printed in your terminal. This confirms Pydantic is reading your .env file correctly.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt.
Confirm you're running the command from the learn-platform-api folder (the folder that contains the app/ directory).
Check that app/__init__.py and app/core/__init__.py exist. Without these files, Python can't import from the app package.

Help me fix my import error.

✔️ Awesome, I've got everything!

Great. Double check you've saved both app/core/config.py and .env.

ⓧ I'd like to double check the full code

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://learnapi:learnapi@localhost:5432/learnplatform"
    redis_url: str = "redis://localhost:6379/0"
    app_name: str = "LearnPlatformAPI"
    debug: bool = False

    model_config = {"env_file": ".env"}


settings = Settings()

DATABASE_URL=postgresql+asyncpg://learnapi:learnapi@localhost:5432/learnplatform
REDIS_URL=redis://localhost:6379/0
APP_NAME=LearnPlatformAPI
DEBUG=true

Your project foundation is locked in. You have a clean directory structure, two database services running in Docker, and a configuration system that pulls settings from environment variables. Next up, you'll define the database models and run your first migration.

Define Models and Run Migrations

Your project infrastructure is ready. Docker containers are running, your dependencies are installed, and your config file knows how to connect to both PostgreSQL and Redis.

Every API needs a way to talk to its database. In this step, you'll build the data layer using SQLAlchemy 2.0's async engine, define your Course and Lesson models, then use Alembic to create the actual database tables.

In this step, get ready to:

Set up the async database engine and session management.
Define Course and Lesson models with a one-to-many relationship.
Initialize Alembic and run the initial migration.

Create the async database module

FastAPI is async-first, so using a synchronous database driver would block the event loop under load. SQLAlchemy 2.0 solves this with native async support through AsyncSession and the asyncpg driver.

Create an empty app/db/__init__.py file to make the db folder a Python package by running:

touch app/db/__init__.py

Create the database configuration file by adding the following code to app/db/database.py:

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from app.core.config import settings

# Create the async engine using the database URL from settings
engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=5,
    max_overflow=10,
    pool_recycle=1800,
    pool_pre_ping=True,
)

# Session factory that produces async sessions
async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


# Base class for all ORM models
class Base(DeclarativeBase):
    pass

What does this code do?

create_async_engine creates a connection pool to your PostgreSQL database. The pool_size and max_overflow settings control how many simultaneous database connections are allowed.

async_sessionmaker is a factory that produces database sessions. The expire_on_commit=False setting is required for async contexts because accessing expired attributes would trigger implicit I/O, which async SQLAlchemy does not allow.

Base is the parent class that all your ORM models will inherit from. SQLAlchemy uses it to track your table definitions.

The engine and session factory are configured, but your API routes still need a way to get a database session for each request.

Add the session dependency function below the Base class in app/db/database.py:

# Dependency that provides an async session per request
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

What does this function do?

get_async_session is a FastAPI dependency. Each API request gets its own database session. If the request succeeds, the session commits the transaction. If an exception occurs, it rolls back to keep the database consistent.

Save app/db/database.py.

Seeing an import error?

Make sure your virtual environment is activated. If you see ModuleNotFoundError: No module named 'sqlalchemy', the packages from Step 1 are not available in your current terminal session.

help me fix an import error with SQLAlchemy in my FastAPI project

✔️ Awesome, I've got everything!

Make sure you saved app/db/database.py before moving on.

ⓧ I'd like to double check the full code

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from app.core.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=5,
    max_overflow=10,
    pool_recycle=1800,
    pool_pre_ping=True,
)

async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Define the Course and Lesson models

Your learning platform needs two core data types: courses and lessons. A course has many lessons, and each lesson belongs to one course. This is a one-to-many relationship.

SQLAlchemy 2.0 uses Python type annotations (Mapped and mapped_column) to define columns. This gives you type safety and better editor autocompletion compared to the older style.

Create the Course model by adding the following code to app/models/course.py:

from datetime import datetime

from sqlalchemy import String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.database import Base


class Course(Base):
    __tablename__ = "courses"

    # Primary key auto-increments by default
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str] = mapped_column(Text, nullable=False)
    instructor: Mapped[str] = mapped_column(String(100), nullable=False)
    # server_default uses the database's NOW() function
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    # One course has many lessons
    lessons: Mapped[list["Lesson"]] = relationship(
        back_populates="course", cascade="all, delete-orphan"
    )

What does this model do?

__tablename__ = "courses" tells SQLAlchemy to create a database table called courses. Each Mapped field becomes a column in that table.

The lessons relationship links courses to lessons. The cascade="all, delete-orphan" setting means deleting a course automatically deletes all its lessons too.

Save app/models/course.py.

Create the Lesson model by adding the following code to app/models/lesson.py:

from datetime import datetime

from sqlalchemy import ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.database import Base


class Lesson(Base):
    __tablename__ = "lessons"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    # Controls the display order of lessons within a course
    order: Mapped[int] = mapped_column(nullable=False)
    # Foreign key linking each lesson to its parent course
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    # Back-reference to the parent course
    course: Mapped["Course"] = relationship(back_populates="lessons")

What does this model do?

The course_id column is a foreign key pointing to courses.id. This is how the database enforces the one-to-many relationship. Every lesson must belong to exactly one course.

The course relationship with back_populates="lessons" creates a two-way link. You can access a lesson's course (lesson.course) or a course's lessons (course.lessons).

Save app/models/lesson.py.

The last piece is the package's __init__.py file, which re-exports both models so other parts of your app can import them easily.

Add the following code to app/models/__init__.py:

from app.models.course import Course
from app.models.lesson import Lesson

__all__ = ["Course", "Lesson"]

Save app/models/__init__.py.
Confirm your model files exist by running:

ls app/models/

You should see __init__.py, course.py, and lesson.py in the output.

Missing a file?

Make sure you created each file inside the app/models/ directory. If you accidentally created them in the project root, move them into the correct folder.

help me fix my model file locations

✔️ Awesome, I've got everything!

Make sure you saved all three model files before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime

from sqlalchemy import String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.database import Base


class Course(Base):
    __tablename__ = "courses"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str] = mapped_column(Text, nullable=False)
    instructor: Mapped[str] = mapped_column(String(100), nullable=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    lessons: Mapped[list["Lesson"]] = relationship(
        back_populates="course", cascade="all, delete-orphan"
    )

from datetime import datetime

from sqlalchemy import ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.database import Base


class Lesson(Base):
    __tablename__ = "lessons"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    order: Mapped[int] = mapped_column(nullable=False)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    course: Mapped["Course"] = relationship(back_populates="lessons")

from app.models.course import Course
from app.models.lesson import Lesson

__all__ = ["Course", "Lesson"]

Initialize Alembic and run the initial migration

Your models define what the database tables should look like in Python. But the actual PostgreSQL database has no tables yet. Alembic bridges this gap by generating migration scripts that create, alter, or drop tables.

Initialize Alembic with the async template by running:

alembic init -t async alembic

What does this command do?

The -t async flag tells Alembic to use its async template. This generates an env.py that works with async database engines instead of synchronous ones. Without this flag, migrations would fail because our project uses asyncpg.

Confirm the files were created by running:

ls alembic/

You should see env.py, script.py.mako, README, and a versions/ directory. You should also see an alembic.ini file in your project root.

Seeing "already exists"?

If you get an error saying the alembic directory already exists, you may have run the command twice. Delete the alembic/ folder and alembic.ini file, then run the init command again.

help me fix my Alembic initialization

Alembic needs two configuration changes before it can generate migrations. First, it needs the database connection URL. Second, it needs to know about your models.

Open alembic.ini in your project root.
Find this line:

sqlalchemy.url = driver://user:pass@localhost/dbname

Replace it with your PostgreSQL connection string, so the line looks like this:

sqlalchemy.url = postgresql+asyncpg://learnapi:learnapi@localhost:5432/learnplatform

Save alembic.ini.

Now you need to tell Alembic about your models so it can auto-detect what tables to create.

Open alembic/env.py.
Find these comment lines near the top of the file:

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = None

Replace that entire block (the four comment lines and the target_metadata = None line) with the following:

# Import models so Alembic can detect them for autogenerate
from app.db.database import Base
from app.models import Course, Lesson

target_metadata = Base.metadata

Why do we import the models here?

Alembic's autogenerate feature compares your Python models against the current database schema. By importing Course and Lesson, those model definitions get registered on Base.metadata. Without this import, Alembic would see an empty metadata object and generate a blank migration.

Save alembic/env.py.

Configuration is done. Time to generate and run the migration.

Generate the initial migration by running:

alembic revision --autogenerate -m "initial"

You should see output that includes Generating ... initial.py ... done. This means Alembic detected your Course and Lesson models and created a migration script in the alembic/versions/ directory.

Seeing "Can't locate revision"?

This error usually means the sqlalchemy.url in alembic.ini is incorrect or the PostgreSQL container is not running. Verify your Docker containers are up by running docker compose ps.

Also check that your alembic/env.py imports match exactly. A typo in the import path will cause Alembic to fail before it connects to the database.

help me fix my Alembic migration error

Apply the migration to create the actual database tables by running:

alembic upgrade head

You should see output like Running upgrade -> <revision_id>, initial. Your PostgreSQL database now has courses and lessons tables.

Connection refused?

If you see Connection refused, your PostgreSQL Docker container may not be running. Check with docker compose ps and restart with docker compose up -d if needed.

help me connect Alembic to my PostgreSQL container

Your data layer is complete. You have an async database engine, two models with a one-to-many relationship, and a migration that created the actual tables in PostgreSQL. Next up, you'll build the repository and service layers that read and write data through these models.

Build the Repository and Service Layers

In the last step, you defined your database models and ran Alembic to create the courses and lessons tables in PostgreSQL.

Now you need a way to actually talk to those tables from your Python code. In this step, you'll build two new layers on top of your models: repositories for database queries and services for business logic.

In this step, get ready to:

Create Pydantic schemas for request validation and response serialization.
Build async repository classes for courses and lessons.
Create service classes that coordinate repository operations.

Create the course repository

The repository pattern gives you a single class that handles all database queries for a particular model. Every read, create, update, and delete goes through the repository.

Why separate repositories from services?

Repositories handle raw database access. Services handle business logic like validation, caching, and coordination. Keeping them separate means you can swap your database or add caching without touching query code.

This is the same Controller > Service > Repository pattern used in Spring and other backend frameworks.

Before writing any repository code, you need to create the new package directories. Your repository and schema files will live in these folders.

Create the new package directories and their __init__.py files by running these commands:

mkdir -p app/schemas app/repositories app/services
touch app/schemas/__init__.py app/repositories/__init__.py app/services/__init__.py

Confirm the directories exist by running:

ls app/

You should see core, db, models, repositories, schemas, and services listed as subdirectories.

The course repository imports Pydantic schemas for its method signatures. You need to create the course schemas first.

Create the file app/schemas/course.py and add the following code:

from datetime import datetime

from pydantic import BaseModel, ConfigDict


# Schema for creating a new course
class CourseCreate(BaseModel):
    title: str
    description: str
    instructor: str


# Schema for partial course updates (all fields optional)
class CourseUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    instructor: str | None = None


# Response schema for a single lesson (used inside CourseResponse)
class LessonResponse(BaseModel):
    id: int
    title: str
    content: str
    order: int
    course_id: int
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)


# Response schema for a course, including its lessons
class CourseResponse(BaseModel):
    id: int
    title: str
    description: str
    instructor: str
    created_at: datetime
    updated_at: datetime
    lessons: list[LessonResponse] = []

    model_config = ConfigDict(from_attributes=True)

What do these schemas do?

CourseCreate and CourseUpdate define what data the API accepts. CourseUpdate makes every field optional so you can update just one field at a time.
CourseResponse and LessonResponse define what data the API returns. The from_attributes=True config tells Pydantic to read data directly from SQLAlchemy model attributes.

Save app/schemas/course.py.

Now you can create the course repository. This class wraps all database operations for the Course model behind clean async methods.

Create the file app/repositories/course_repo.py and add the following code:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models.course import Course
from app.schemas.course import CourseCreate, CourseUpdate


class CourseRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_all(self) -> list[Course]:
        # Eager-load lessons with selectinload to avoid async lazy-load errors
        stmt = select(Course).options(selectinload(Course.lessons)).order_by(Course.id)
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def get_by_id(self, course_id: int) -> Course | None:
        stmt = (
            select(Course)
            .options(selectinload(Course.lessons))
            .where(Course.id == course_id)
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, data: CourseCreate) -> Course:
        # Convert Pydantic schema to model fields
        course = Course(**data.model_dump())
        self.session.add(course)
        await self.session.flush()
        await self.session.refresh(course, attribute_names=["lessons"])
        return course

    async def update(self, course: Course, data: CourseUpdate) -> Course:
        # Only apply fields that were explicitly set in the request
        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(course, key, value)
        await self.session.flush()
        await self.session.refresh(course, attribute_names=["lessons"])
        return course

    async def delete(self, course: Course) -> None:
        await self.session.delete(course)
        await self.session.flush()

What does this code do?

Each method takes an AsyncSession injected through __init__. The session handles the database connection for each request.
selectinload(Course.lessons) tells SQLAlchemy to load related lessons in a second query. This avoids the lazy-loading problem in async contexts where accessing course.lessons would trigger an implicit database call that async does not allow.
flush() sends changes to the database without committing. The get_async_session dependency you built in Step 2 handles the final commit or rollback.

Save app/repositories/course_repo.py.
Verify the import chain works by running:

python -c "from app.repositories.course_repo import CourseRepository; print('Import OK')"

You should see Import OK printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure you created the __init__.py files in app/schemas/ and app/repositories/. Python needs these files to recognize the folders as packages.
Check that you saved app/schemas/course.py before running the import check. The repository imports from that file.
Confirm your virtual environment is activated. Your prompt should show the venv name.

Why can't Python find my CourseRepository?

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime

from pydantic import BaseModel, ConfigDict


class CourseCreate(BaseModel):
    title: str
    description: str
    instructor: str


class CourseUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    instructor: str | None = None


class LessonResponse(BaseModel):
    id: int
    title: str
    content: str
    order: int
    course_id: int
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)


class CourseResponse(BaseModel):
    id: int
    title: str
    description: str
    instructor: str
    created_at: datetime
    updated_at: datetime
    lessons: list[LessonResponse] = []

    model_config = ConfigDict(from_attributes=True)

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models.course import Course
from app.schemas.course import CourseCreate, CourseUpdate


class CourseRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_all(self) -> list[Course]:
        stmt = select(Course).options(selectinload(Course.lessons)).order_by(Course.id)
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def get_by_id(self, course_id: int) -> Course | None:
        stmt = (
            select(Course)
            .options(selectinload(Course.lessons))
            .where(Course.id == course_id)
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, data: CourseCreate) -> Course:
        course = Course(**data.model_dump())
        self.session.add(course)
        await self.session.flush()
        await self.session.refresh(course, attribute_names=["lessons"])
        return course

    async def update(self, course: Course, data: CourseUpdate) -> Course:
        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(course, key, value)
        await self.session.flush()
        await self.session.refresh(course, attribute_names=["lessons"])
        return course

    async def delete(self, course: Course) -> None:
        await self.session.delete(course)
        await self.session.flush()

Create the lesson repository

The lesson repository follows the same pattern as the course repository. The key difference is that lessons are always scoped to a specific course.

Create the file app/schemas/lesson.py and add the following code:

from datetime import datetime

from pydantic import BaseModel, ConfigDict


# Schema for creating a new lesson
class LessonCreate(BaseModel):
    title: str
    content: str
    order: int


# Schema for partial lesson updates
class LessonUpdate(BaseModel):
    title: str | None = None
    content: str | None = None
    order: int | None = None


# Response schema for lesson data
class LessonResponse(BaseModel):
    id: int
    title: str
    content: str
    order: int
    course_id: int
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)

What does this code do?

LessonCreate requires a title, content, and order field. The order field controls the sequence of lessons within a course. LessonUpdate makes all fields optional so you can patch one field without providing the rest.

Save app/schemas/lesson.py.

Create the file app/repositories/lesson_repo.py and add the following code:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.lesson import Lesson
from app.schemas.lesson import LessonCreate, LessonUpdate


class LessonRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_course(self, course_id: int) -> list[Lesson]:
        # Fetch all lessons for a specific course, sorted by display order
        stmt = (
            select(Lesson)
            .where(Lesson.course_id == course_id)
            .order_by(Lesson.order)
        )
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def get_by_id(self, lesson_id: int) -> Lesson | None:
        stmt = select(Lesson).where(Lesson.id == lesson_id)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, course_id: int, data: LessonCreate) -> Lesson:
        # Attach the lesson to a course via course_id
        lesson = Lesson(course_id=course_id, **data.model_dump())
        self.session.add(lesson)
        await self.session.flush()
        await self.session.refresh(lesson)
        return lesson

    async def update(self, lesson: Lesson, data: LessonUpdate) -> Lesson:
        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(lesson, key, value)
        await self.session.flush()
        await self.session.refresh(lesson)
        return lesson

    async def delete(self, lesson: Lesson) -> None:
        await self.session.delete(lesson)
        await self.session.flush()

What's happening here?

get_by_course() filters lessons by course_id and orders them by the order column. This ensures lessons always come back in the sequence the instructor intended.
The create() method takes course_id as a separate parameter and injects it into the Lesson constructor. The course_id comes from the URL path, not from the request body.
Notice there is no selectinload here. Lessons do not eagerly load their parent course because the API response only needs the lesson data, not the full course object.

Save app/repositories/lesson_repo.py.
Verify the lesson repository imports correctly by running:

python -c "from app.repositories.lesson_repo import LessonRepository; print('Import OK')"

You should see Import OK in your terminal.

Seeing an ImportError?

Double check that app/schemas/lesson.py is saved. The lesson repository imports LessonCreate and LessonUpdate from it.
Check for typos in the import paths. The file is lesson_repo.py with an underscore, not lessonrepo.py.

Why is my LessonRepository import failing?

✔️ Awesome, I've got everything!

Great. Make sure both the schema and repository files are saved.

ⓧ I'd like to double check the full code

from datetime import datetime

from pydantic import BaseModel, ConfigDict


class LessonCreate(BaseModel):
    title: str
    content: str
    order: int


class LessonUpdate(BaseModel):
    title: str | None = None
    content: str | None = None
    order: int | None = None


class LessonResponse(BaseModel):
    id: int
    title: str
    content: str
    order: int
    course_id: int
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.lesson import Lesson
from app.schemas.lesson import LessonCreate, LessonUpdate


class LessonRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_course(self, course_id: int) -> list[Lesson]:
        stmt = (
            select(Lesson)
            .where(Lesson.course_id == course_id)
            .order_by(Lesson.order)
        )
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def get_by_id(self, lesson_id: int) -> Lesson | None:
        stmt = select(Lesson).where(Lesson.id == lesson_id)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, course_id: int, data: LessonCreate) -> Lesson:
        lesson = Lesson(course_id=course_id, **data.model_dump())
        self.session.add(lesson)
        await self.session.flush()
        await self.session.refresh(lesson)
        return lesson

    async def update(self, lesson: Lesson, data: LessonUpdate) -> Lesson:
        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(lesson, key, value)
        await self.session.flush()
        await self.session.refresh(lesson)
        return lesson

    async def delete(self, lesson: Lesson) -> None:
        await self.session.delete(lesson)
        await self.session.flush()

Create the service layers

Services sit between your API routes and the repositories. Right now the services simply pass calls through to the repository. In the next step, you will add Redis caching logic at this layer.

Create the file app/services/course_service.py and add the following code:

from app.models.course import Course
from app.repositories.course_repo import CourseRepository
from app.schemas.course import CourseCreate, CourseUpdate


class CourseService:
    def __init__(self, repo: CourseRepository):
        self.repo = repo

    async def get_all(self) -> list[Course]:
        return await self.repo.get_all()

    async def get_by_id(self, course_id: int) -> Course | None:
        return await self.repo.get_by_id(course_id)

    async def create(self, data: CourseCreate) -> Course:
        return await self.repo.create(data)

    async def update(self, course_id: int, data: CourseUpdate) -> Course | None:
        # Look up the course first, return None if not found
        course = await self.repo.get_by_id(course_id)
        if not course:
            return None
        return await self.repo.update(course, data)

    async def delete(self, course_id: int) -> bool:
        # Return False if the course doesn't exist
        course = await self.repo.get_by_id(course_id)
        if not course:
            return False
        await self.repo.delete(course)
        return True

What does this code do?

The service wraps every repository method. Read methods (get_all, get_by_id) pass straight through to the repository for now.
Write methods (update, delete) add a lookup step. They fetch the course first and return None or False if the course does not exist. The router layer will convert these into HTTP 404 responses.

Save app/services/course_service.py.

Create the file app/services/lesson_service.py and add the following code:

from app.models.lesson import Lesson
from app.repositories.lesson_repo import LessonRepository
from app.schemas.lesson import LessonCreate, LessonUpdate


class LessonService:
    def __init__(self, repo: LessonRepository):
        self.repo = repo

    async def get_by_course(self, course_id: int) -> list[Lesson]:
        return await self.repo.get_by_course(course_id)

    async def get_by_id(self, lesson_id: int) -> Lesson | None:
        return await self.repo.get_by_id(lesson_id)

    async def create(self, course_id: int, data: LessonCreate) -> Lesson:
        return await self.repo.create(course_id, data)

    async def update(self, lesson_id: int, data: LessonUpdate) -> Lesson | None:
        # Look up the lesson first, return None if not found
        lesson = await self.repo.get_by_id(lesson_id)
        if not lesson:
            return None
        return await self.repo.update(lesson, data)

    async def delete(self, lesson_id: int) -> bool:
        # Return False if the lesson doesn't exist
        lesson = await self.repo.get_by_id(lesson_id)
        if not lesson:
            return False
        await self.repo.delete(lesson)
        return True

What does this code do?

LessonService follows the same delegation pattern as CourseService. Each method wraps the corresponding repository call, with write operations adding a not-found check. The lesson service does not receive a Redis client because lesson caching is handled through course-level cache invalidation.

Save app/services/lesson_service.py.
Verify both service imports work by running:

python -c "from app.services.course_service import CourseService; from app.services.lesson_service import LessonService; print('All services OK')"

You should see All services OK in your terminal.

Seeing a SyntaxError or ImportError?

Check that app/services/__init__.py exists. You created it with the touch command at the start of this step.
Make sure you are running the command from the project root directory (the folder containing app/).

Why can't Python find my service classes?

✔️ Awesome, I've got everything!

Great. Make sure both service files are saved before moving on.

ⓧ I'd like to double check the full code

from app.models.course import Course
from app.repositories.course_repo import CourseRepository
from app.schemas.course import CourseCreate, CourseUpdate


class CourseService:
    def __init__(self, repo: CourseRepository):
        self.repo = repo

    async def get_all(self) -> list[Course]:
        return await self.repo.get_all()

    async def get_by_id(self, course_id: int) -> Course | None:
        return await self.repo.get_by_id(course_id)

    async def create(self, data: CourseCreate) -> Course:
        return await self.repo.create(data)

    async def update(self, course_id: int, data: CourseUpdate) -> Course | None:
        course = await self.repo.get_by_id(course_id)
        if not course:
            return None
        return await self.repo.update(course, data)

    async def delete(self, course_id: int) -> bool:
        course = await self.repo.get_by_id(course_id)
        if not course:
            return False
        await self.repo.delete(course)
        return True

from app.models.lesson import Lesson
from app.repositories.lesson_repo import LessonRepository
from app.schemas.lesson import LessonCreate, LessonUpdate


class LessonService:
    def __init__(self, repo: LessonRepository):
        self.repo = repo

    async def get_by_course(self, course_id: int) -> list[Lesson]:
        return await self.repo.get_by_course(course_id)

    async def get_by_id(self, lesson_id: int) -> Lesson | None:
        return await self.repo.get_by_id(lesson_id)

    async def create(self, course_id: int, data: LessonCreate) -> Lesson:
        return await self.repo.create(course_id, data)

    async def update(self, lesson_id: int, data: LessonUpdate) -> Lesson | None:
        lesson = await self.repo.get_by_id(lesson_id)
        if not lesson:
            return None
        return await self.repo.update(lesson, data)

    async def delete(self, lesson_id: int) -> bool:
        lesson = await self.repo.get_by_id(lesson_id)
        if not lesson:
            return False
        await self.repo.delete(lesson)
        return True

Your repository and service layers are in place. The entire data access stack is wired up: services call repositories, and repositories talk to PostgreSQL through async sessions. Next up, you'll add a Redis caching layer to the course service so reads hit the cache instead of the database.

Add Redis Caching with the Cache-Aside Pattern

Your repository and service layers give the API clean database access. But every single request still hits PostgreSQL directly, even for data that hasn't changed since the last query.

In this step, you'll add a Redis caching layer that sits between the service and the database. The API will check Redis first and only query PostgreSQL when the cache is empty or expired.

In this step, get ready to:

Create the Redis client module with connection pooling and lifecycle functions.
Update the course service with cache-aside reads and write-through invalidation.

Create the Redis client module

The Redis container is already running from your docker-compose.yml. Now you need a Python module that connects to it, verifies the connection on startup, and provides a dependency your services can use.

Create a new file called redis.py inside your app/core/ folder.
Paste the following code into app/core/redis.py:

import redis.asyncio as aioredis

from app.core.config import settings

# Global connection pool and client, initialized at app startup
redis_pool: aioredis.ConnectionPool | None = None
redis_client: aioredis.Redis | None = None


async def init_redis() -> None:
    """Create the Redis connection pool and verify connectivity."""
    global redis_pool, redis_client
    redis_pool = aioredis.ConnectionPool.from_url(
        settings.redis_url,
        max_connections=10,
        decode_responses=True,
    )
    redis_client = aioredis.Redis(connection_pool=redis_pool)
    # Ping Redis to confirm it's reachable
    await redis_client.ping()


async def close_redis() -> None:
    """Gracefully shut down the Redis connection."""
    if redis_client:
        await redis_client.close()
    if redis_pool:
        await redis_pool.disconnect()


def get_redis() -> aioredis.Redis:
    """Dependency that provides the Redis client to services."""
    if redis_client is None:
        raise RuntimeError("Redis client is not initialized")
    return redis_client

What does this code do?

redis_pool and redis_client are module-level globals. They start as None and get initialized when the app starts up.
init_redis() creates a connection pool from the REDIS_URL in your .env file, then pings Redis to confirm it is reachable.
close_redis() cleanly shuts down the client and pool when the app stops.
get_redis() is a FastAPI dependency. Any service that needs Redis calls this to get the active client.

Why a connection pool instead of a single connection?

A connection pool keeps multiple connections to Redis ready. When several API requests arrive at the same time, each one grabs an available connection from the pool instead of waiting for a single shared connection.

decode_responses=True tells the pool to return strings from Redis instead of raw bytes. This saves you from manually decoding every cached value.

Save app/core/redis.py.

You won't see a change yet. The next step wires init_redis() and close_redis() into the FastAPI lifespan so they run on startup and shutdown.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

import redis.asyncio as aioredis

from app.core.config import settings

redis_pool: aioredis.ConnectionPool | None = None
redis_client: aioredis.Redis | None = None


async def init_redis() -> None:
    global redis_pool, redis_client
    redis_pool = aioredis.ConnectionPool.from_url(
        settings.redis_url,
        max_connections=10,
        decode_responses=True,
    )
    redis_client = aioredis.Redis(connection_pool=redis_pool)
    await redis_client.ping()


async def close_redis() -> None:
    if redis_client:
        await redis_client.close()
    if redis_pool:
        await redis_pool.disconnect()


def get_redis() -> aioredis.Redis:
    if redis_client is None:
        raise RuntimeError("Redis client is not initialized")
    return redis_client

What is the cache-aside pattern?

Cache-aside means the application manages the cache explicitly. On a read, the service checks Redis first. If the data is there (a cache hit), it returns the cached value immediately. If not (a cache miss), it queries the database, stores the result in Redis with a time-to-live (TTL), and then returns it.

On writes (create, update, delete), the service performs the database operation first, then deletes the stale cache keys. The next read repopulates the cache with fresh data.

Why cache-aside instead of write-through?

Write-through updates the cache on every write, which gives higher consistency but adds complexity. Cache-aside only populates the cache on reads and invalidates on writes. For a read-heavy API like a learning platform, cache-aside is simpler and a better fit.

Update the course service with caching

Your CourseService currently delegates every call straight to the repository. You'll replace it with a version that checks Redis first on reads and invalidates cached keys on writes.

Open app/services/course_service.py.
Select all the existing code and delete it.
Paste the new imports, cache constants, and updated __init__:

import json

import redis.asyncio as aioredis

from app.models.course import Course
from app.repositories.course_repo import CourseRepository
from app.schemas.course import CourseCreate, CourseResponse, CourseUpdate

# Cache key patterns and TTL (time-to-live) in seconds
COURSE_LIST_KEY = "courses:all"
COURSE_KEY_PREFIX = "courses:"
CACHE_TTL = 300


class CourseService:
    def __init__(self, repo: CourseRepository, redis: aioredis.Redis):
        self.repo = repo
        self.redis = redis

What does this code do?

COURSE_LIST_KEY is the Redis key for the full list of courses. COURSE_KEY_PREFIX is used to build keys for individual courses (e.g. courses:5).
CACHE_TTL = 300 sets cached data to expire after 300 seconds (5 minutes). After that, the next read triggers a fresh database query.
The __init__ now accepts both a repository and a Redis client. The service needs both to implement the cache-aside logic.

Add the get_all method inside the CourseService class, directly below self.redis = redis:

    async def get_all(self) -> list[Course]:
        # Check Redis for a cached course list
        cached = await self.redis.get(COURSE_LIST_KEY)
        if cached:
            courses_data = json.loads(cached)
            return [Course(**c) for c in courses_data]

        # Cache miss: query the database
        courses = await self.repo.get_all()
        # Serialize to JSON and store in Redis with a TTL
        courses_json = json.dumps(
            [CourseResponse.model_validate(c).model_dump(mode="json") for c in courses]
        )
        await self.redis.setex(COURSE_LIST_KEY, CACHE_TTL, courses_json)
        return courses

What does this code do?

This is the cache-aside pattern in action. The method first asks Redis for the courses:all key. If it exists, the cached JSON is deserialized directly into Course objects without touching the database.

On a cache miss, it queries PostgreSQL through the repository, converts each course to JSON using the Pydantic CourseResponse schema, and stores the result in Redis with setex (SET with EXpiry). The TTL ensures stale data eventually refreshes itself.

Add the get_by_id method directly below get_all:

    async def get_by_id(self, course_id: int) -> Course | None:
        # Build a unique cache key for this course
        cache_key = f"{COURSE_KEY_PREFIX}{course_id}"
        cached = await self.redis.get(cache_key)
        if cached:
            return None  # Signal to use cached data in the router

        # Cache miss: query the database
        course = await self.repo.get_by_id(course_id)
        if course:
            # Serialize and cache the individual course
            course_json = CourseResponse.model_validate(course).model_dump(mode="json")
            await self.redis.setex(
                cache_key, CACHE_TTL, json.dumps(course_json)
            )
        return course

What does this code do?

The same cache-aside pattern, but for a single course. The cache key includes the course ID (e.g. courses:5), so each course has its own dedicated cache entry. If the course is found in the database, it gets cached for future requests.

Add the write methods (create, update, delete) and the private invalidation helpers below get_by_id:

    async def create(self, data: CourseCreate) -> Course:
        course = await self.repo.create(data)
        # New course changes the list, so invalidate the list cache
        await self._invalidate_list_cache()
        return course

    async def update(self, course_id: int, data: CourseUpdate) -> Course | None:
        course = await self.repo.get_by_id(course_id)
        if not course:
            return None
        updated = await self.repo.update(course, data)
        # Invalidate both the individual course cache and the list
        await self._invalidate_course_cache(course_id)
        return updated

    async def delete(self, course_id: int) -> bool:
        course = await self.repo.get_by_id(course_id)
        if not course:
            return False
        await self.repo.delete(course)
        # Invalidate both the individual course cache and the list
        await self._invalidate_course_cache(course_id)
        return True

    async def _invalidate_list_cache(self) -> None:
        await self.redis.delete(COURSE_LIST_KEY)

    async def _invalidate_course_cache(self, course_id: int) -> None:
        await self.redis.delete(f"{COURSE_KEY_PREFIX}{course_id}")
        await self.redis.delete(COURSE_LIST_KEY)

What does this code do?

create() saves the new course to the database, then deletes the courses:all cache key. The next list request will re-query the database and cache the updated list.
update() and delete() invalidate both the individual course key and the list key. Deleting just the individual key would leave stale data in the list cache.
The two private _invalidate helpers keep the invalidation logic in one place. If you add more cache keys later, you only update these methods.

Save app/services/course_service.py.

You won't see a visible change yet. The next step creates the API routes and main.py entry point, where you'll be able to test the caching in action through the Swagger docs.

Seeing import errors in your editor?

Make sure your virtual environment is activated. Your editor's Python interpreter should point to the venv folder inside your project directory.
Check that redis is installed. Run pip list | grep redis in your terminal. You should see redis 7.4.0.
Verify that app/core/redis.py exists. The course service imports CourseResponse from your schemas, so also confirm app/schemas/course.py is in place.

Help me debug import errors in my course service

✔️ Awesome, I've got everything!

Great. Double check you've saved app/services/course_service.py before moving on.

ⓧ I'd like to double check the full code

import json

import redis.asyncio as aioredis

from app.models.course import Course
from app.repositories.course_repo import CourseRepository
from app.schemas.course import CourseCreate, CourseResponse, CourseUpdate

COURSE_LIST_KEY = "courses:all"
COURSE_KEY_PREFIX = "courses:"
CACHE_TTL = 300


class CourseService:
    def __init__(self, repo: CourseRepository, redis: aioredis.Redis):
        self.repo = repo
        self.redis = redis

    async def get_all(self) -> list[Course]:
        cached = await self.redis.get(COURSE_LIST_KEY)
        if cached:
            courses_data = json.loads(cached)
            return [Course(**c) for c in courses_data]

        courses = await self.repo.get_all()
        courses_json = json.dumps(
            [CourseResponse.model_validate(c).model_dump(mode="json") for c in courses]
        )
        await self.redis.setex(COURSE_LIST_KEY, CACHE_TTL, courses_json)
        return courses

    async def get_by_id(self, course_id: int) -> Course | None:
        cache_key = f"{COURSE_KEY_PREFIX}{course_id}"
        cached = await self.redis.get(cache_key)
        if cached:
            return None  # Signal to use cached data in the router

        course = await self.repo.get_by_id(course_id)
        if course:
            course_json = CourseResponse.model_validate(course).model_dump(mode="json")
            await self.redis.setex(
                cache_key, CACHE_TTL, json.dumps(course_json)
            )
        return course

    async def create(self, data: CourseCreate) -> Course:
        course = await self.repo.create(data)
        await self._invalidate_list_cache()
        return course

    async def update(self, course_id: int, data: CourseUpdate) -> Course | None:
        course = await self.repo.get_by_id(course_id)
        if not course:
            return None
        updated = await self.repo.update(course, data)
        await self._invalidate_course_cache(course_id)
        return updated

    async def delete(self, course_id: int) -> bool:
        course = await self.repo.get_by_id(course_id)
        if not course:
            return False
        await self.repo.delete(course)
        await self._invalidate_course_cache(course_id)
        return True

    async def _invalidate_list_cache(self) -> None:
        await self.redis.delete(COURSE_LIST_KEY)

    async def _invalidate_course_cache(self, course_id: int) -> None:
        await self.redis.delete(f"{COURSE_KEY_PREFIX}{course_id}")
        await self.redis.delete(COURSE_LIST_KEY)

Your caching layer is in place. Next up, you'll create the API routes and wire everything together into a running FastAPI server that you can test through the browser.

Wire Up the API Routes and Test

You have a complete backend stack: models, repositories, services, and a Redis caching layer. The missing piece is the HTTP interface that lets clients actually talk to your API.

In this step, you will create the FastAPI routers that define your endpoints, build the application entry point that ties everything together, and test the full CRUD flow through the interactive Swagger docs.

In this step, get ready to:

Create course and lesson API routers with full CRUD endpoints.
Build the FastAPI application entry point with Redis lifespan management.
Test the complete API flow using the interactive Swagger docs.

Create the API route files

Routers are the top layer of the layered architecture. They define HTTP endpoints and delegate the actual work to the service layer below.

Create the app/api/v1/ directory structure and its empty __init__.py files by running these commands:

mkdir -p app/api/v1
touch app/api/__init__.py app/api/v1/__init__.py

Confirm the files exist by running:

ls app/api/ app/api/v1/

You should see __init__.py listed under both directories.

Create a new file called app/api/v1/courses.py and add the following code:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.redis import get_redis
from app.db.database import get_async_session
from app.repositories.course_repo import CourseRepository
from app.schemas.course import CourseCreate, CourseResponse, CourseUpdate
from app.services.course_service import CourseService

# Create a router that groups all course-related endpoints
router = APIRouter(prefix="/courses", tags=["Courses"])


# Dependency factory: builds a CourseService with a fresh session and Redis client
def get_course_service(
    session: AsyncSession = Depends(get_async_session),
    redis=Depends(get_redis),
) -> CourseService:
    repo = CourseRepository(session)
    return CourseService(repo, redis)


# List all courses
@router.get("/", response_model=list[CourseResponse])
async def list_courses(service: CourseService = Depends(get_course_service)):
    return await service.get_all()


# Get a single course by ID
@router.get("/{course_id}", response_model=CourseResponse)
async def get_course(
    course_id: int, service: CourseService = Depends(get_course_service)
):
    course = await service.get_by_id(course_id)
    if not course:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Course not found"
        )
    return course


# Create a new course
@router.post("/", response_model=CourseResponse, status_code=status.HTTP_201_CREATED)
async def create_course(
    data: CourseCreate, service: CourseService = Depends(get_course_service)
):
    return await service.create(data)


# Update an existing course
@router.put("/{course_id}", response_model=CourseResponse)
async def update_course(
    course_id: int,
    data: CourseUpdate,
    service: CourseService = Depends(get_course_service),
):
    course = await service.update(course_id, data)
    if not course:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Course not found"
        )
    return course


# Delete a course
@router.delete("/{course_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_course(
    course_id: int, service: CourseService = Depends(get_course_service)
):
    deleted = await service.delete(course_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Course not found"
        )

What does this code do?

This file defines five CRUD endpoints for courses, all grouped under a single router.

get_course_service() is a dependency factory. FastAPI calls it automatically for every request, injecting a fresh database session and the Redis client. This is how the layered architecture stays wired together without hard-coded connections.
Each endpoint delegates to the service layer and never touches the database directly. The router's only job is to handle HTTP concerns: status codes, error responses, and request/response schemas.
POST returns 201 Created and DELETE returns 204 No Content. These are the standard HTTP status codes for those operations.

Save app/api/v1/courses.py.

You won't see a change yet. The next file creates the lesson router, and then we will wire both into the application.

Create a new file called app/api/v1/lessons.py and add the following code:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.database import get_async_session
from app.repositories.lesson_repo import LessonRepository
from app.schemas.lesson import LessonCreate, LessonResponse, LessonUpdate
from app.services.lesson_service import LessonService

# Lessons are nested under a course: /courses/{course_id}/lessons
router = APIRouter(prefix="/courses/{course_id}/lessons", tags=["Lessons"])


# Dependency factory: builds a LessonService with a fresh session
def get_lesson_service(
    session: AsyncSession = Depends(get_async_session),
) -> LessonService:
    repo = LessonRepository(session)
    return LessonService(repo)


# List all lessons for a specific course
@router.get("/", response_model=list[LessonResponse])
async def list_lessons(
    course_id: int, service: LessonService = Depends(get_lesson_service)
):
    return await service.get_by_course(course_id)


# Create a new lesson under a course
@router.post("/", response_model=LessonResponse, status_code=status.HTTP_201_CREATED)
async def create_lesson(
    course_id: int,
    data: LessonCreate,
    service: LessonService = Depends(get_lesson_service),
):
    return await service.create(course_id, data)


# Update an existing lesson
@router.put("/{lesson_id}", response_model=LessonResponse)
async def update_lesson(
    course_id: int,
    lesson_id: int,
    data: LessonUpdate,
    service: LessonService = Depends(get_lesson_service),
):
    lesson = await service.update(lesson_id, data)
    if not lesson:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Lesson not found"
        )
    return lesson


# Delete a lesson
@router.delete("/{lesson_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_lesson(
    course_id: int,
    lesson_id: int,
    service: LessonService = Depends(get_lesson_service),
):
    deleted = await service.delete(lesson_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Lesson not found"
        )

What does this code do?

The lesson router follows the same pattern as the course router, with one key difference.

The prefix /courses/{course_id}/lessons makes lessons a nested resource. Every lesson endpoint requires a course_id in the URL, which scopes operations to a specific course.
The lesson service has no Redis dependency because lesson caching is not implemented in this project. Only course reads hit the cache.

Save app/api/v1/lessons.py.

You won't see a change yet. The next substep creates the application entry point that mounts both routers.

Seeing import errors in your editor?

Your editor may show red squiggles under the imports. This is normal. The files reference each other, and until all files exist, the editor cannot resolve every import path.

Make sure your virtual environment is selected as the Python interpreter. In VS Code, check the bottom-right corner for the correct environment.

Help me fix import errors in my FastAPI project

Create the application entry point

The main.py file is where every component comes together. It creates the FastAPI app, manages the Redis connection lifecycle, and mounts the routers under a versioned prefix.

Create a new file called app/main.py and add the following code:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api.v1.courses import router as courses_router
from app.api.v1.lessons import router as lessons_router
from app.core.config import settings
from app.core.redis import close_redis, init_redis


# Lifespan manages startup and shutdown events for the app
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize the Redis connection pool and verify connectivity
    await init_redis()
    yield
    # Shutdown: cleanly close the Redis connection pool
    await close_redis()


app = FastAPI(
    title=settings.app_name,
    lifespan=lifespan,
)

# Mount both routers under the /api/v1 prefix
app.include_router(courses_router, prefix="/api/v1")
app.include_router(lessons_router, prefix="/api/v1")


# Simple health check endpoint
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

What does this code do?

The lifespan context manager runs init_redis() when the server starts and close_redis() when it shuts down. This ensures the Redis connection pool is always properly managed.
include_router() mounts the routers under /api/v1. Combined with each router's own prefix, the final paths become /api/v1/courses and /api/v1/courses/{course_id}/lessons.
The /health endpoint is a quick way to confirm the server is running without hitting the database or cache.

Save app/main.py.
Make sure your Docker containers are still running by executing:

docker compose ps

You should see both the db and redis services in a running state. If they are not running, start them with docker compose up -d.

Start the FastAPI development server by running:

uvicorn app.main:app --reload

You should see output including Uvicorn running on http://127.0.0.1:8000 in your terminal.

What does --reload do?

The --reload flag tells Uvicorn to watch your files and automatically restart the server whenever you save a change. This is a development convenience. You would not use it in production.

Server not starting?

If you see ModuleNotFoundError, make sure your virtual environment is activated. You should see the environment name in your terminal prompt.

If you see Connection refused errors related to PostgreSQL or Redis, confirm your Docker containers are running with docker compose ps.

Help me troubleshoot my FastAPI server startup error

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the full code for each new file. Compare your files against these.

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.redis import get_redis
from app.db.database import get_async_session
from app.repositories.course_repo import CourseRepository
from app.schemas.course import CourseCreate, CourseResponse, CourseUpdate
from app.services.course_service import CourseService

router = APIRouter(prefix="/courses", tags=["Courses"])


def get_course_service(
    session: AsyncSession = Depends(get_async_session),
    redis=Depends(get_redis),
) -> CourseService:
    repo = CourseRepository(session)
    return CourseService(repo, redis)


@router.get("/", response_model=list[CourseResponse])
async def list_courses(service: CourseService = Depends(get_course_service)):
    return await service.get_all()


@router.get("/{course_id}", response_model=CourseResponse)
async def get_course(
    course_id: int, service: CourseService = Depends(get_course_service)
):
    course = await service.get_by_id(course_id)
    if not course:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Course not found"
        )
    return course


@router.post("/", response_model=CourseResponse, status_code=status.HTTP_201_CREATED)
async def create_course(
    data: CourseCreate, service: CourseService = Depends(get_course_service)
):
    return await service.create(data)


@router.put("/{course_id}", response_model=CourseResponse)
async def update_course(
    course_id: int,
    data: CourseUpdate,
    service: CourseService = Depends(get_course_service),
):
    course = await service.update(course_id, data)
    if not course:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Course not found"
        )
    return course


@router.delete("/{course_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_course(
    course_id: int, service: CourseService = Depends(get_course_service)
):
    deleted = await service.delete(course_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Course not found"
        )

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.database import get_async_session
from app.repositories.lesson_repo import LessonRepository
from app.schemas.lesson import LessonCreate, LessonResponse, LessonUpdate
from app.services.lesson_service import LessonService

router = APIRouter(prefix="/courses/{course_id}/lessons", tags=["Lessons"])


def get_lesson_service(
    session: AsyncSession = Depends(get_async_session),
) -> LessonService:
    repo = LessonRepository(session)
    return LessonService(repo)


@router.get("/", response_model=list[LessonResponse])
async def list_lessons(
    course_id: int, service: LessonService = Depends(get_lesson_service)
):
    return await service.get_by_course(course_id)


@router.post("/", response_model=LessonResponse, status_code=status.HTTP_201_CREATED)
async def create_lesson(
    course_id: int,
    data: LessonCreate,
    service: LessonService = Depends(get_lesson_service),
):
    return await service.create(course_id, data)


@router.put("/{lesson_id}", response_model=LessonResponse)
async def update_lesson(
    course_id: int,
    lesson_id: int,
    data: LessonUpdate,
    service: LessonService = Depends(get_lesson_service),
):
    lesson = await service.update(lesson_id, data)
    if not lesson:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Lesson not found"
        )
    return lesson


@router.delete("/{lesson_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_lesson(
    course_id: int,
    lesson_id: int,
    service: LessonService = Depends(get_lesson_service),
):
    deleted = await service.delete(lesson_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Lesson not found"
        )

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api.v1.courses import router as courses_router
from app.api.v1.lessons import router as lessons_router
from app.core.config import settings
from app.core.redis import close_redis, init_redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_redis()
    yield
    await close_redis()


app = FastAPI(
    title=settings.app_name,
    lifespan=lifespan,
)

app.include_router(courses_router, prefix="/api/v1")
app.include_router(lessons_router, prefix="/api/v1")


@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Test the complete API flow

FastAPI automatically generates interactive API documentation from your router definitions and Pydantic schemas. You can use this to test every endpoint without writing a single line of client code.

Open your browser and go to http://localhost:8000/docs.

You should see the Swagger UI with two groups of endpoints: Courses and Lessons.

Expand the POST /api/v1/courses/ endpoint.
Click Try it out.
Replace the request body with this JSON:

{
  "title": "Python for Beginners",
  "description": "Learn Python from scratch with hands-on projects",
  "instructor": "Alice Smith"
}

Click Execute.

You should see a 201 response with the created course, including auto-generated id, created_at, and updated_at fields.

Expand the POST /api/v1/courses/{course_id}/lessons/ endpoint.
Click Try it out.
Enter 1 in the course_id field.
Replace the request body with this JSON:

{
  "title": "Variables and Data Types",
  "content": "In this lesson, you will learn about Python variables and basic data types.",
  "order": 1
}

Click Execute.

You should see a 201 response with the lesson data, including course_id: 1 linking it to the course you just created.

Expand the GET /api/v1/courses/ endpoint.
Click Try it out, then Execute.

You should see the course with its nested lesson inside the lessons array. This first request hit the database and cached the result in Redis.

Verify the cache is working by opening a second terminal and running:

docker compose exec redis redis-cli KEYS '*'

You should see courses:all in the output. This is the cache key your service created when you listed courses.

Now test cache invalidation. Go back to the Swagger UI and expand PUT /api/v1/courses/{course_id}.
Click Try it out.
Enter 1 for the course_id.
Replace the request body with this JSON:

{
  "title": "Python for Beginners - Updated"
}

Click Execute.

The response should show the updated title. The update also invalidated the cache.

Confirm the cache was cleared by running the Redis command again in your second terminal:

docker compose exec redis redis-cli KEYS '*'

The output should now be (empty array) or show no courses: keys. The update triggered _invalidate_course_cache(), which deleted both the individual course key and the list key.

Run the GET /api/v1/courses/ request again in Swagger.

The response should show the updated course title. This request hit the database (since the cache was empty) and re-cached the fresh data.

What just happened with the cache?

You just saw the full cache-aside cycle in action. The first GET request populated the cache. The PUT request invalidated it. The next GET request rebuilt the cache with the updated data.

This pattern guarantees that clients always see fresh data after a write, while still getting the performance benefits of caching on subsequent reads.

Getting 404 or 500 errors?

If you see a 500 Internal Server Error, check the terminal running Uvicorn for the full traceback. Common causes include missing Alembic migrations (run alembic upgrade head) or Docker containers that are not running.

If the KEYS command shows no cached keys after a GET request, verify that your .env file has REDIS_URL=redis://localhost:6379/0.

Help me debug errors in my FastAPI learning platform API

Secret mission

Cursor-Based Pagination and a Leaderboard Endpoint

Your API handles basic CRUD with caching, but real-world APIs need pagination for large datasets and analytics endpoints for quick insights. In this secret mission, you'll add cursor-based pagination to the course listing and build a Redis-cached leaderboard that ranks courses by lesson count.

Before you go

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally using Docker containers, so there are no ongoing cloud costs.

Resources you used:

PostgreSQL 16 Docker container (database).
Redis 7 Docker container (cache).
Docker volume postgres_data (persisted database files).
Python virtual environment (.venv folder).
Project directory with application code, Alembic migrations, and configuration files.

Keep everything running

No action needed. Choose this if you plan to keep developing your API or experimenting with new endpoints.

Your PostgreSQL and Redis containers stay running in Docker and consume minimal resources while idle.
Your database tables and any data you created through the API will persist across restarts.

Pause - I'll come back to this later

Stop the Docker containers to free up memory without losing your data. You can restart them later.

Open your terminal and navigate to your project folder.
Stop the containers by running:

docker compose stop

When you're ready to resume, start the containers again by running:

docker compose start

Your database data and Redis cache will still be there when you come back.

Delete - I don't want to use this again

Remove all Docker containers, volumes, and project files to start completely fresh.

Open your terminal and navigate to your project folder.
Stop and remove the containers along with all data volumes by running:

docker compose down -v

What does the -v flag do?

The -v flag removes named volumes declared in your docker-compose.yml. Without it, the postgres_data volume would persist on disk even after the containers are gone.

Optionally, delete the project folder and its virtual environment by running:

cd ~/Desktop
rm -rf learnplatform-api

This removes all application code, the virtual environment, and Alembic migration files.

Mission accomplished

That's a Wrap!

Nice work! You just built a production-style backend API with FastAPI, async PostgreSQL, and Redis caching.

You've learned how to:

Built a layered architecture with routers, services, and repositories, wired together through FastAPI dependency injection.
Set up async PostgreSQL with SQLAlchemy 2.0 Mapped type annotations, ran migrations with Alembic, and created full CRUD endpoints with Pydantic v2 request and response validation.
Implemented the cache-aside pattern with Redis, including TTL-based expiration and write-through invalidation on every create, update, and delete.
Added cursor-based pagination to the course listing and a Redis-cached leaderboard endpoint ranking courses by lesson count.

Ready to quiz yourself?