Intermediate

Building a REST API with Python 3

In this course you will learn how to build a complete Python project from scratch. You will use the excellent and very popular FastAPI framework to build a REST API that serves data from...

Versions covered: Python 3.10+, Poetry, FastAPI 0.103+, SQLAlchemy 2.0+

Table of Contents

  1. Course Overview
  2. Project start
  1. Creating the data model
  1. Make content accessible via an API
  1. Reading contents from filesystem
  1. Complete project structure

1. Course Overview

Welcome to this course: Building a REST API with Python 3. The instructor is Reindert-Jan Ekker, a Python developer and trainer at his own company called Code Sensei.

In this course you will learn how to build a complete Python project from scratch. You will use the excellent and very popular FastAPI framework to build a REST API that serves data from a database, using SQLAlchemy for the data model.

The general approach is to follow best practices to create a project that resembles real-world software development as closely as possible.

Main topics covered

  • Creating a REST API with FastAPI
  • Data modeling with SQLAlchemy
  • Complete unit testing with pytest
  • Working with frontmatter files using regular expressions and YAML

At the end of the course, you will know how to work with a complete and realistic Python project.

2. Project start

2.1 Introduction and prerequisites

This course was created with the displayed versions of Python, Poetry, FastAPI and SQLAlchemy. The information in this course applies to these specific versions.

Target audience

This course is intended for intermediate level Python developers. Here is an overview of the subjects you must master as prerequisites:

Python language and standard library:

  • Basics of Object Oriented Programming (OOP), including the use of inheritance and properties
  • pathlib for working with text files
  • Regular expressions to match text patterns

Code quality:

  • Flake8 to check code style
  • mypy for type annotations (type hints)
  • Poetry for project setup and dependency management

Tests:

  • pytest for unit testing

External libraries:

  • SQLAlchemy for database access
  • requests to test the REST API in unit tests
  • FastAPI to build the API
  • Pydantic for data validation
  • Uvicorn as HTTP server

2.2 Presentation of the GloboTicket project

The client is GloboTicket — a company that sells tickets to events like concerts.

Current problem

All of their systems currently use a legacy system where data is stored in two places:

  1. Database: contains only part of the data — product codes, dates, prices
  2. Frontmatter files (represented by a YAML file icon): contain other information such as the event name, artist, description, etc.

To obtain all the information for a product, you must consult the database and find the corresponding frontmatter file, then combine the two.

Currently the website and other systems (like billing) connect to these two different data sources to bring everything together.

Desired solution

The client wants us to build a REST API which:

  • Connects to both data sources
  • Presents a unified interface

The client plans to create a new website, mobile application, and other systems in the future. It wants to be independent of data storage technology. All systems will only talk to the REST API.

This will allow them to change the way data is stored (e.g. migrate everything to one database or use cloud storage) without having to change the front-end systems. Only the API will need to be updated.

2.3 Getting started — get the source code

The source code for the project is in a GitHub repository. To work in parallel with the course (which is strongly recommended for better learning), you must collect this project:

  • Clone the repository with Git
  • Or download the zip file

To manage dependencies, we use Poetry. If you don’t have it yet, follow the link to the installation instructions.

Project initialization

The project was created with the command:

poetry new globoticket

Poetry created:

  • A package called globoticket (client name)
  • A tests folder (empty, containing only a __init__.py file)
  • In the globoticket package: a __init__.py file and an api.py file added manually
  • The pyproject.toml file
  • An empty README.md file

There are also data files provided at startup (the events.db database, product files, and static HTML files).

To start at the correct version of the project:

git clone <url-du-repo>
git checkout project-start

2.4 Start the server

runserver.py

The runserver.py script imports Uvicorn (an HTTP server — a library that can serve websites) and tells it to run app from globoticket.api:

"""
runserver.py
---

Run this script to start the globoticket app.
"""
import uvicorn

def main():
    uvicorn.run(
        "globoticket.api:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
    )

if __name__ == "__main__":
    main()

Important parameters:

  • host="0.0.0.0": listen for local connections (the four zeros mean “listen for local connections”)
  • port=8000: the TCP port to listen on
  • reload=True: automatically reloads the application when you modify the Python code

globoticket/api.py (initial release)

"""
api.py
---

The REST Api for the Globoticket events database.
"""

from fastapi import FastAPI
from starlette.staticfiles import StaticFiles
from pathlib import Path

app = FastAPI()

PROJECT_ROOT = Path(__file__).parent.parent
app.mount("/", StaticFiles(directory=PROJECT_ROOT / "static", html=True))

FastAPI is imported and an app object is created. It is the application object which represents the REST API. This is what is referenced in the Uvicorn server startup (globoticket.api then the app object).

app.mount("/", StaticFiles(...)): The server must directly serve certain files from the file system. These files are served as is — we call them static files (as opposed to dynamic content generated by code). The PROJECT_ROOT/static directory contains HTML files representing what the final product might look like when there is a website using our REST API.

To start the server:

python runserver.py
# ou, si pas dans l'environnement virtuel actif :
poetry run python runserver.py

2.5 Open project in PyCharm

When we open the project in PyCharm, we should see the message “No Python interpreter configured” with the option to configure a Poetry environment from pyproject.toml. Or, if the environment has already been created with Poetry on the command line, you can activate it directly.

Configuration steps in PyCharm

  1. Click at the bottom right on “No interpreter”
  2. Select “Add New Interpreter” → “Add Local Interpreter”
  3. In the window, choose “Poetry Environment” (if not visible: update PyCharm)
  4. Specify the Poetry installation path

To find the path to Poetry:

# Si installé avec pipx :
pipx list
# => indique le répertoire d'installation (ex. ~/.local/bin)

# Si installé avec pip :
pip show poetry

If the Poetry environment has already been created (with poetry install): select “Existing environment” — PyCharm will automatically detect it.

Once configured, you can launch the project from the integrated terminal:

# Dans le terminal PyCharm (environnement actif automatiquement)
python runserver.py

2.6 Open project in VS Code

In Visual Studio Code, when opening the project, the Python interpreter may not be correctly selected.

Configuration steps in VS Code

  1. Click at the bottom right on the name of the interpreter (ex. “No interpreter”)
  2. A list of available environments appears, including Poetry environments
  3. Select the GloboTicket Poetry environment

If VS Code does not detect it automatically:

# Dans le dossier du projet, demander à Poetry le chemin de l'environnement :
poetry env info
# => affiche le chemin de l'exécutable du venv

Then in VS Code: click on the interpreter → “Enter interpreter path” → paste the path.

2.7 The pre-commit framework

One last step to complete the project configuration: install the pre-commit framework.

# Avec poetry run (fonctionne même en dehors de l'environnement actif) :
poetry run pre-commit install

# Ou directement si dans l'environnement actif :
pre-commit install

This command creates a Git hook — a piece of code that runs before any commit.

Configuration in .pre-commit-config.yaml

# See https://pre-commit.com for more information
repos:
  - repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
      - id: black
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.2.0
    hooks:
      - id: mypy
  - repo: https://github.com/PyCQA/flake8
    rev: 6.0.0
    hooks:
      - id: flake8
        args: ["--max-line-length=88", "--extend-ignore=E203"]
  - repo: https://github.com/PyCQA/isort
    rev: 5.12.0
    hooks:
    - id: isort
      args: [ "--profile=black" ]

During a commit, pre-commit will automatically execute:

  • black: formats files
  • mypy: checks type annotations
  • Flake8: checks the code style (code smells)
  • isort: sorts import instructions

Example of operation

If we insert bad code into api.py — a poorly formatted function with incorrect spaces and a bad type annotation — and try to commit:

  1. The first time, pre-commit takes a little while because it installs black, mypy, etc.
  2. black indicates that it has reformatted the file
  3. mypy reports the problem with the type annotation (eg: return type str but the function returns None)
  4. isort sorted the imports

Commit is rejected due to mypy error. You need to fix the problem, then re-commit.

Note: Files modified by black and isort are in the “modified not staged” state. You must add them with git add before re-committing.

3. Creating the data model

3.1 Module Overview

In this module, we will write code to read and write to the customer provided database, using SQLAlchemy.

We will also make the URL of the database configurable to be able to run the project with different databases (production, development, etc.). For these configuration parameters, we will use the python-dotenv library — the de facto standard for this type of configuration.

The customer shared the SQL code used to create the existing tables in the database:

  • Table category: an ID and a name
  • Table event: an ID, a product code, a date, a price, and a foreign key to category

3.2 Configure the database

SQL schema of the database (database.ddl)

create table category
(
    id   INTEGER not null
        primary key,
    name VARCHAR not null
        unique
);

create table event
(
    id           INTEGER not null
        primary key,
    product_code VARCHAR not null
        unique,
    date         DATE    not null,
    price        NUMERIC not null,
    category_id  INTEGER not null
        references category
);

The events.db database currently contains 3 events and 3 categories.

database.py (initial version with hardcoded URL)

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///events.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    echo=True,
)
SessionLocal = sessionmaker(bind=engine)

Explanation:

  • SQLALCHEMY_DATABASE_URL: String telling SQLAlchemy how to connect. For SQLite, it’s simply a file name. For Postgres, it would look like postgres://server-name
  • engine: represents the database connection
  • connect_args={"check_same_thread": False}: specific to SQLite, allows multi-threaded access
  • echo=True: SQLAlchemy will display all executed SQL queries
  • SessionLocal: factory to create database sessions

3.3 Create the data model for categories

We create a new models.py file. The data model for category is:

  • id: primary key of type int
  • name: string with uniqueness constraint
import decimal
from datetime import date

from sqlalchemy import ForeignKey
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    declarative_base,
    mapped_column,
    relationship,
)

Base: DeclarativeBase = declarative_base()

class DBCategory(Base):
    __tablename__ = "category"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    events: Mapped[list["DBEvent"]] = relationship(back_populates="category")

Naming convention: We prefix DB model classes with DB (e.g. DBCategory) to distinguish them from other classes of the same name that might appear later (like Pydantic schemas).

To test that it works (temporary test code):

# Dans models.py, ajouter temporairement :
session = SessionLocal()
results = session.execute(select(DBCategory)).scalars()
print("\n".join(category.name for category in results))

Execution:

python globoticket/models.py
# ou :
poetry run python globoticket/models.py

SQLAlchemy executes an SQL query and the category names appear in the output.

3.4 Create the data model for events

The model for the event table:

class DBEvent(Base):
    __tablename__ = "event"

    id: Mapped[int] = mapped_column(primary_key=True)
    product_code: Mapped[str] = mapped_column(unique=True)
    date: Mapped[date]
    price: Mapped[decimal.Decimal]
    category_id: Mapped[int] = mapped_column(ForeignKey("category.id"))
    category: Mapped["DBCategory"] = relationship(back_populates="events")

Important points:

  • date: type date imported from datetime
  • price: type decimal.Decimal — good practice for monetary data

Why Decimal for the price? If we represent money as a floating point number (float), we can get rounding errors during calculations. With integers (int), we cannot represent cents. The best practice for monetary data in a database is to use an infinite precision type. In Python, it is Decimal. In SQLite, this is a NUMERIC field.

  • category_id: foreign key to category.id
  • category: additional attribute mapped as a SQLAlchemy relationship. This allows access to the DBCategory object directly from a DBEvent

3.5 Create a relationship between models

Lazy loading vs joined

Looking at the output of SQLAlchemy during a query that accesses the event categories, we see 4 queries to print 3 events:

  1. A query to select all events
  2. A separate query for each category

This is because SQLAlchemy relationships are lazy by default: they only load when needed. When selecting events, SQLAlchemy only loads the event table. Only when event.category is accessed does SQLAlchemy execute a query to load the category.

To force immediate loading (joined loading):

category: Mapped["DBCategory"] = relationship(back_populates="events", lazy="joined")

With lazy="joined", only one query with a JOIN is executed. The disadvantage: the JOIN is always performed, even if we don’t need the categories. After reflection, we return to lazy loading by default (by removing this argument).

The two-way relationship with back_populates

With relationship(back_populates="events") in DBEvent and relationship(back_populates="category") in DBCategory, the relationship is bidirectional:

  • From an event, you can access its category: event.category
  • From a category, you can access all its events: category.events

These are two sides of the same relationship. The foreign key in the DBEvent class points to a single category, but a category can have multiple events.

3.6 Make the database configurable with Dotenv

In database.py, the database URL is hardcoded, which is not a good practice. On a production server or on another developer’s machine, the file might be elsewhere, or one might want to use a different type of database.

Installing python-dotenv

poetry add python-dotenv

Note: The package was already required by other packages like Uvicorn, so it was already installed. The command only adds the main dependency in pyproject.toml.

database.py (final version with dotenv)

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = os.getenv("SQLALCHEMY_DATABASE_URL")

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    echo=True,
)
SessionLocal = sessionmaker(bind=engine)

os.getenv("SQLALCHEMY_DATABASE_URL") reads an environment variable — a variable that comes not from the Python code, but from the program’s execution environment (the shell).

In the terminal (to test):

export SQLALCHEMY_DATABASE_URL="sqlite:///events.db"
python globoticket/database.py

But it is more convenient to store these values ​​in an .env file.

.env file

# Usually you don't add this file to git
SQLALCHEMY_DATABASE_URL="sqlite:///events.db"

At the top of database.py, we call load_dotenv():

from dotenv import load_dotenv

load_dotenv()

load_dotenv() reads the .env file and loads environment variables from this file. So os.getenv("SQLALCHEMY_DATABASE_URL") will read the value from the .env file.

Best practice: We generally not commit the .env file to Git because it contains the local configuration, which differs from one developer to another or from one environment to another. For this course, we still commit it for simplicity.

3.7 Commit to version control

Thinking about code quality

Docstrings: The names DBEvent and DBCategory are sufficient in themselves. The field definitions are very readable. Adding a docstring saying “this class represents an event” would be unnecessary. Same goes for models.py — the module name is descriptive enough.

Unit tests: The functions written don’t really contain any custom logic — it’s mostly SQLAlchemy configuration. Writing tests that verify that we can execute SQL queries would simply be testing SQLAlchemy itself, not our own code. There is therefore no need to write unit tests at this stage.

Resolving mypy error

During commit, mypy complains about the line:

Base = declarative_base()

mypy cannot determine the type. The correction is to add a type annotation:

from sqlalchemy.orm import DeclarativeBase, declarative_base

Base: DeclarativeBase = declarative_base()

Resolving Flake8 Warnings

Flake8 reports several unused imports in files. This is the part that requires manual cleanup work: removing unused imports in each file.

In PyCharm (Mac): Ctrl+Alt+O to optimize imports.

4. Make content accessible via an API

4.1 Create an API endpoint

General structure of a FastAPI endpoint

@app.get("/hello")
def hello() -> str:
    return "hello"

The hello function returns a string "hello" and is decorated with @app.get. Decoration is what turns an ordinary function into an API endpoint (or, as FastAPI calls it, an operation). An operation always needs an assigned URL — here /hello.

Behavior: When a client requests the /hello URL, FastAPI calls this function. It is the function’s responsibility to determine what the server should return in response. Here it simply returns the string "hello".

This can be checked in the browser by accessing http://localhost:8000/hello.

4.2 DB session injection

The get_session function

from sqlalchemy.orm import Session

def get_session() -> Session:
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

This is a generator function (it uses yield). The session is created, then yield to provide it to the calling code. When the call is completed (whether there is an error or not), the finally block closes the session.

Dependency Injection with Annotated and Depends

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException

@app.get("/event/{id}", response_model=Event)
def get_event(id: int, db: Annotated[Session, Depends(get_session)]) -> DBEvent:
    """Retrieve a single event by id. Returns status 404 if event is not found."""
    event = get_dbevent(id, db)
    if event is None:
        raise HTTPException(status_code=404, detail=f"No product with id {id}")
    return event

Annotated is a Python feature not often used outside of FastAPI. Available since Python 3.9 in the typing module, it allows you to annotate a function argument with more than just its type.

The call to Annotated takes two arguments:

  1. Session — the classic type annotation: db is an object of class Session
  2. Depends(get_session) — FastAPI specific annotation

Depends(get_session) tells FastAPI to create the db object by calling the get_session function. This is called dependency injection: FastAPI takes care of calling get_session and passing the resulting session to the get_event function.

4.3 Take an argument from the URL

@app.get("/event/{id}", response_model=Event)
def get_event(id: int, db: Annotated[Session, Depends(get_session)]) -> DBEvent:

In the decorator, {id} in the URL is a path parameter. It specifies part of the URL that will be used for the function’s id argument.

Example: if we access /event/1, the 1 is used as the id parameter. The event with id=1 is retrieved from the database and returned to the client.

Advantages of type annotation for parameters:

  • Automatic validation: if you enter something that is not an integer, FastAPI returns a validation error (eg: "id is not a valid integer")
  • Automatic documentation: FastAPI uses type annotation to document parameters

Error case — 404 Not Found:

if event is None:
    raise HTTPException(status_code=404, detail=f"No product with id {id}")

When the event does not exist, we throw an HTTPException with the code 404.

4.4 FastAPI documentation

FastAPI automatically generates documentation for API endpoints. It can be accessed at the URL /docs.

What we find there:

  • The list of all endpoints (e.g.: get_event, name taken from the name of the function)
  • The operation type (GET)
  • The endpoint URL
  • Parameters (e.g.: id, type integer)
  • A “Try it out” button to directly test the endpoint as a test client

Adding a docstring improves the documentation:

def get_event(id: int, db: ...) -> DBEvent:
    """Retrieve a single event by id. Returns status 404 if event is not found."""

When refreshing the docs, this docstring appears in the documentation.

FastAPI generates comprehensive and user-friendly documentation for your API, which is one of the reasons for its great popularity.

Current limitation: There is no actual documentation of the returned JSON structure. To remedy this, we must formally define what the endpoint returns — this is the role of schemas.

4.5 Define a response pattern

schemas.py — The Event class (Pydantic)

import datetime
import decimal

from pydantic import BaseModel

class Event(BaseModel):
    id: int
    date: datetime.date
    price: decimal.Decimal

Class Event inherits from BaseModel (imported from Pydantic). Such a class is similar to a standard Python dataclass — it’s a quick way to define a class with fields.

What you get for free with BaseModel:

  • A __init__ method that validates its inputs
  • Type validation at instantiation

Demonstration in the Python console:

from globoticket.schemas import Event
from datetime import date

# Cela échoue car "2024-not-a-date" n'est pas une date valide :
e = Event(id=1, date="2024-not-a-date", price=10.50)
# => Pydantic lève une ValidationError

# Version correcte :
e = Event(id=1, date=date.today(), price=10.50)
# => fonctionne

Difference between DBEvent (SQLAlchemy) and Event (Pydantic)

| Class | Library | Role | |--------|------------||------| | DBEvent | SQLAlchemy | Defines the table structure in the database | | Event | Pydantic | Defines the structure of JSON data returned by the API |

Event is used to define what will be sent by the API. Such a formal definition of a data structure is called a schema — hence the name schemas.py.

Use schema in endpoint

from globoticket.schemas import Event

@app.get("/event/{id}", response_model=Event)
def get_event(id: int, db: ...) -> DBEvent:

The response_model=Event parameter in the decorator tells FastAPI to use this pattern for the response. In the docs the JSON structure of the response is now documented. The diagrams also validate the answer.

4.6 An endpoint that returns a list of events

Exercise then solution

@app.get("/event/", response_model=list[Event])
def get_all_events(db: Annotated[Session, Depends(get_session)]) -> list[DBEvent]:
    return get_all_dbevents(db)

The get_all_events function has the same db argument as get_event. The return type is slightly different: a list of DBEvent objects. The response_model says that we return a list of Event schemas.

Endpoint order — important point: FastAPI always processes the first matching endpoint. If we put get_all_events before get_event, all requests starting with /event would be handled by get_all_events, making get_event inaccessible. You must therefore put get_event first. All /event/{something} requests will be handled by get_event. Only /event/ requests with nothing after them will be processed by get_all_events.

4.7 Make the API testable

Extracting to crud.py

To make the code more testable, we extract the data access logic into a separate crud.py file (CRUD = Create, Read, Update, Delete):

from sqlalchemy import select
from sqlalchemy.orm import Session

from globoticket.models import DBEvent

def get_dbevent(id: int, db: Session) -> DBEvent | None:
    return db.get(DBEvent, id)

def get_all_dbevents(db: Session) -> list[DBEvent]:
    return db.execute(select(DBEvent)).scalars()

Advantage: Instead of mocking the call to SQLAlchemy, we can now mock the get_dbevent function, which is simpler.

We update api.py to use CRUD functions:

from globoticket.crud import get_all_dbevents, get_dbevent

@app.get("/event/{id}", response_model=Event)
def get_event(id: int, db: Annotated[Session, Depends(get_session)]) -> DBEvent:
    """Retrieve a single event by id. Returns status 404 if event is not found."""
    event = get_dbevent(id, db)
    if event is None:
        raise HTTPException(status_code=404, detail=f"No product with id {id}")
    return event

@app.get("/event/", response_model=list[Event])
def get_all_events(db: Annotated[Session, Depends(get_session)]) -> list[DBEvent]:
    return get_all_dbevents(db)

4.8 Unit tests with mocking

test_get_event.py (version with mocks)

from unittest.mock import patch

import pytest
from fastapi import HTTPException

from globoticket.api import get_event
from globoticket.models import DBEvent

@patch("globoticket.api.get_dbevent", return_value=DBEvent(product_code="Test"))
def test_get_event(mock_get_dbevent):
    """Return the event found in the database."""
    assert get_event(id=25, db="Fake db") is mock_get_dbevent.return_value
    mock_get_dbevent.assert_called_with(25, "Fake db")

@patch("globoticket.api.get_dbevent", return_value=None)
def test_get_event_404(_):
    """Raise HTTPException when no event is found."""
    with pytest.raises(HTTPException):
        get_event(id=0, db=None)

Explanation of the decorator @patch

@patch("globoticket.api.get_dbevent") replaces the get_dbevent identifier in the globoticket.api module with a mock object. We use globoticket.api.get_dbevent (and not globoticket.crud.get_dbevent) because we replace the name as it exists in the API module (which imports the function), not where it is defined.

The mock is passed as an argument to the test function (mock_get_dbevent). We can:

  • Set a return value: return_value=DBEvent(product_code="Test")
  • Check how it was called: mock_get_dbevent.assert_called_with(25, "Fake db")

The test verifies:

  1. test_get_event: what get_dbevent returns must also be what get_event returns
  2. test_get_event_404: when get_dbevent returns None, get_event should throw an HTTPException

Advantage of mocking: We do not use the real database. We only test the logic of get_event in isolation.

4.9 Integration testing with TestClient

CustomerTest Concept

FastAPI includes its own TestClient, allowing you to write integration tests that simulate real HTTP calls. It comes from Starlette (which is included with FastAPI). Behind the scenes it uses HTTPX (an alternative to requests) but exposes the same interface.

from starlette.testclient import TestClient
from globoticket.api import app

client = TestClient(app)
response = client.get("/event/1")

This is not a unit test but an integration test. The function under test runs in the context of FastAPI and makes real queries to the (test) database.

Benefits of Integration Testing:

  • Test how everything works together
  • Detect Pydantic validation errors
  • Detect errors in model classes or CRUD module

conftest.py — Configuring the test database

"""
conftest.py
---

Fixtures for pytest.
"""
from datetime import date
from pathlib import Path
from unittest.mock import patch

import pytest
import sqlalchemy
from sqlalchemy.orm import Session, sessionmaker
from starlette.testclient import TestClient

from globoticket.api import app, get_session
from globoticket.models import Base, DBCategory, DBEvent

# Créer une base de données en mémoire pour les tests
test_db = sqlalchemy.create_engine(
    "sqlite+pysqlite:///:memory:",
    connect_args={"check_same_thread": False},
    echo=True,
)

test_sessionmaker = sessionmaker(bind=test_db)

def setup_test_db():
    """Create tables and test data in test db."""
    Base.metadata.create_all(bind=test_db)
    session = Session(test_db)
    cat = DBCategory(name="t")
    ev = DBEvent(product_code="123456", price=5.50, date=date(2024, 1, 1), category=cat)
    session.add(cat)
    session.add(ev)
    session.commit()

def get_test_session():
    """Create a db session for a single test.
    After the test: close the session and drop all tables."""
    setup_test_db()
    session = test_sessionmaker()
    try:
        yield session
    finally:
        session.close()
        Base.metadata.drop_all(bind=test_db)

@pytest.fixture()
def client():
    """Create a TestClient that uses the test database."""
    # See: https://fastapi.tiangolo.com/advanced/testing-database/
    app.dependency_overrides[get_session] = get_test_session
    yield TestClient(app)
    del app.dependency_overrides[get_session]

Key points:

  • We create an SQLite database in memory ("sqlite+pysqlite:///:memory:") — it disappears at the end of the tests
  • app.dependency_overrides[get_session] = get_test_session: replace dependency get_session with the test version. FastAPI will call our get_test_session instead of get_session to create the DB session
  • After the tests, we remove the replacement: del app.dependency_overrides[get_session]
  • Tables are created and dropped for each test (create_all and drop_all)

test_get_event.py (integration tests)

def test_client_get_event_1(client):
    """Call get_event using the client, retrieving event 1"""
    response = client.get("/event/1")
    assert response.status_code == 200
    assert response.json() == {
        "date": "2024-01-01",
        "id": 1,
        "price": "5.5000000000",
    }

def test_client_get_event_404(client):
    """Call get_event using the client, with a non-existing id"""
    response = client.get("/event/100")
    assert response.status_code == 404

5. Reading contents from file system

5.1 Introduction to frontmatter files

In the previous module, we implemented endpoints to retrieve event data from the database. But some data is still missing: artist, image, name, etc., because these properties are not stored in the database but on the file system.

Structure of the product_info folder

The additional data is located in the product_info folder of the project. It contains subfolders for different music genres. Each subfolder can contain frontmatter files.

Example frontmatter file (product_info/rock/23534643.yml):

---
name: "Album Presentation: Batteries Included"
artist: Guido and the Snakes
published: t
image: guido.png
---
Guido van Rossum and his band of Pythonistas run through all the wonderful lines they have written
for their new Batteries Included Album.

Structure of a frontmatter file

A frontmatter file is composed of two parts separated by a line with three hyphens (---):

  1. First part (YAML): event metadata — name, artist, image file name, publishing status, etc. It’s a YAML format — a dictionary of key-value pairs (like a Python dict)
  2. Second part (text): textual content (description) after the separation line

5.2 Searching for files with pathlib

find_frontmatter_file function

Objective: Given a product code, recursively search the product_info folder for a file whose name matches this product code and ends with .yml.

We practice Test-Driven Development (TDD): we start with the tests that fail, then we implement the function.

Unit tests (test_find_frontmatter_files.py):

from pathlib import Path

import pytest

from globoticket.frontmatter import find_frontmatter_file

FM_PATH = Path(__file__).parent / "product_info"

def test_find_file_toplevel():
    """Locate a yaml file at toplevel directory"""
    assert find_frontmatter_file("111222", FM_PATH) == FM_PATH / "111222.yml"

def test_find_file_subdirs():
    """Locate a yaml file in subdirectory"""
    assert find_frontmatter_file("123456", FM_PATH) == FM_PATH / "hiphop" / "123456.yml"
    assert find_frontmatter_file("654321", FM_PATH) == FM_PATH / "reggae" / "654321.yml"

def test_find_nosuchfile():
    """Raise correct exception when file does not exist"""
    with pytest.raises(FileNotFoundError):
        find_frontmatter_file("000000", FM_PATH)

Implementation of find_frontmatter_file:

def find_frontmatter_file(product_code: str, frontmatter_dir: Path) -> Path:
    """Find a file named `product_code`.yml in frontmatter_dir
    and its subdirectories.

    Raises FileNotFoundError if no file is found for this product_code."""
    matches = list(frontmatter_dir.glob(f"**/{product_code}.yml"))
    if not matches:
        raise FileNotFoundError(
            f"Cannot find yaml file for product {product_code} in {frontmatter_dir}"
        )
    return matches[0]

Explanation:

  • frontmatter_dir.glob(f"**/{product_code}.yml"): uses pathlib’s glob
  • **: match any subfolder (recursively)
  • /{product_code}.yml: searches for a file with this name and extension
  • If no file found → FileNotFoundError
  • Returns the first result found

5.3 Parsing files with Regex and pyyaml

Format analysis

The frontmatter file has the following structure:

---
<données YAML>
---
<contenu texte>

We will use a regular expression to divide the file into parts, then PyYAML to parse the YAML part.

Custom exception type

class InvalidFrontmatterError(Exception):
    pass

We create a custom exception type by inheriting from Exception and leaving the body empty (pass).

The regular expression FRONTMATTER_REGEX

FRONTMATTER_REGEX = re.compile(
    r"---\n(?P<yaml>.*?)---\n(?P<content>.*)",
    re.DOTALL,
)

Explanation of named groups:

  • (?P<yaml>.*?): group named yaml — captures YAML data (the ? makes the match non-greedy)
  • (?P<content>.*): group named content — captures content text
  • re.DOTALL: the . also corresponds to line breaks

parse_frontmatter function

def parse_frontmatter(frontmatter: str) -> dict[str, Any]:
    """Return the contents of frontmatter as a dict.

    Raises an InvalidFrontmatterError if anything is wrong."""

    match = re.match(FRONTMATTER_REGEX, frontmatter)
    if not match:
        raise InvalidFrontmatterError("Invalid file structure")

    try:
        data = yaml.load(match.group("yaml"), Loader=yaml.FullLoader)
        data["content"] = match.group("content") or ""
    except (yaml.YAMLError, TypeError) as e:
        raise InvalidFrontmatterError("Invalid YAML (should be a dict)") from e

    return data

Two things that can go wrong:

  1. The YAML is invalid → yaml.load raises yaml.YAMLError
  2. The YAML is valid but is not a dict (YAML allows other types) → TypeError because we are trying to assign data["content"] to a non-dict

Unit tests for parse_frontmatter

import pytest

from globoticket.frontmatter import InvalidFrontmatterError, parse_frontmatter

def test_parse_frontmatter():
    """Parse a frontmatter file."""
    fm = parse_frontmatter(
        """---
name: Live at Folsom Prison
artist: Johnny Cash
published: false
---
Sold out"""
    )
    assert fm == {
        "content": "Sold out",
        "name": "Live at Folsom Prison",
        "artist": "Johnny Cash",
        "published": False,
    }

def test_parse_empty():
    """An empty file causes an error. So do files with only dashes in them."""
    for s in ["", "----", "---\n---"]:
        with pytest.raises(InvalidFrontmatterError):
            parse_frontmatter(s)

def test_parse_content_only():
    """A file without YAML raises an error"""
    with pytest.raises(InvalidFrontmatterError):
        parse_frontmatter(
            """---
---
Test content"""
        )

def test_parse_yaml_only():
    """Parse a file with YAML but empty content."""
    fm = parse_frontmatter(
        """---
name: Live at Folsom Prison
artist: Johnny Cash
published: false
---
"""
    )
    assert fm == {
        "content": "",
        "name": "Live at Folsom Prison",
        "artist": "Johnny Cash",
        "published": False,
    }

def test_parse_yaml_invalid():
    """Invalid YAML is not allowed."""
    with pytest.raises(InvalidFrontmatterError):
        parse_frontmatter(
            """---
    - name: test
    - artist: testartist
    - published: true
    ---
    Test content!"""
        )

5.4 Add file contents to data model

Constant FRONTMATTER_DIRECTORY

FRONTMATTER_DIRECTORY = Path(__file__).parent.parent / "product_info"

get_frontmatter function

def get_frontmatter(product_code: str) -> dict[str, Any]:
    """Read the frontmatter for event with given product_code,
    and return the properties as a dict."""
    frontmatter = find_frontmatter_file(product_code, FRONTMATTER_DIRECTORY).read_text()
    return parse_frontmatter(frontmatter)

This function combines find_frontmatter_file and parse_frontmatter: it finds the file, reads its contents, and parses it.

Trying with __init__ (does not work with SQLAlchemy)

We could think of adding a __init__ method to DBEvent:

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    frontmatter = get_frontmatter(self.product_code)
    for k, v in frontmatter.items():
        if not hasattr(self, k):
            setattr(self, k, v)

But it doesn’t work: SQLAlchemy does not call __init__ when it reads a record from the database. It uses its own mechanisms to reconstruct objects.

Solution with @reconstructor

from sqlalchemy.orm import reconstructor

class DBEvent(Base):
    __tablename__ = "event"

    id: Mapped[int] = mapped_column(primary_key=True)
    product_code: Mapped[str] = mapped_column(unique=True)
    date: Mapped[date]
    price: Mapped[decimal.Decimal]
    category_id: Mapped[int] = mapped_column(ForeignKey("category.id"))
    category: Mapped["DBCategory"] = relationship(back_populates="events")

    @reconstructor
    def _get_frontmatter(self):
        frontmatter = get_frontmatter(self.product_code)
        for k, v in frontmatter.items():
            if not hasattr(self, k):
                setattr(self, k, v)

The @reconstructor decorator tells SQLAlchemy to execute this method when it rebuilds an object from the database. So the frontmatter attributes are dynamically added to the DBEvent object via setattr.

Logic:

  • For each key-value in the frontmatter dictionary, if the attribute does not yet exist on the object (e.g. artist), we add it with setattr
  • If the attribute already exists (e.g. price which comes from the DB), we do not overwrite it

Pydantic schema update

Now that the DBEvent object has additional fields, we update schemas.py:

import datetime
import decimal

from pydantic import BaseModel

class Event(BaseModel):
    id: int
    date: datetime.date
    price: decimal.Decimal
    artist: str
    name: str
    content: str
    image: str

5.5 Fix unit tests

Problem

Integration tests fail because the constant FRONTMATTER_DIRECTORY in frontmatter.py points to the real product_info folder of the project — not the test folder.

First approach (by test) — not recommended

@patch(
    "globoticket.frontmatter.FRONTMATTER_DIRECTORY",
    new=Path(__file__).parent / "product_info",
)
def test_client_get_event_1(client):
    ...

Although this works, if you have many similar tests, you have to copy and paste this decorator everywhere. It’s not clean.

Best approach — autouse fixture in conftest.py

@pytest.fixture(autouse=True, scope="session")
def test_frontmatter():
    """Override the location of frontmatter files."""
    with patch(
        "globoticket.frontmatter.FRONTMATTER_DIRECTORY",
        new=Path(__file__).parent / "product_info",
    ):
        yield

autouse=True: this fixture runs automatically without the tests needing to explicitly request it.

scope="session": Runs once for the entire test session.

The with patch(...): block with yield encapsulates all tests in the context of the patch — all tests will run with FRONTMATTER_DIRECTORY pointing to the tests folder.

full final conftest.py (Module 5)

"""
conftest.py
---

Fixtures for pytest.
"""
from datetime import date
from pathlib import Path
from unittest.mock import patch

import pytest
import sqlalchemy
from sqlalchemy.orm import Session, sessionmaker
from starlette.testclient import TestClient

from globoticket.api import app, get_session
from globoticket.models import Base, DBCategory, DBEvent

test_db = sqlalchemy.create_engine(
    "sqlite+pysqlite:///:memory:",
    connect_args={"check_same_thread": False},
    echo=True,
)

test_sessionmaker = sessionmaker(bind=test_db)

def setup_test_db():
    """Create tables and test data in test db."""
    Base.metadata.create_all(bind=test_db)
    session = Session(test_db)
    cat = DBCategory(name="t")
    ev = DBEvent(product_code="123456", price=5.50, date=date(2024, 1, 1), category=cat)
    session.add(cat)
    session.add(ev)
    session.commit()

def get_test_session():
    """Create a db session for a single test.
    After the test: close the session and drop all tables."""
    setup_test_db()
    session = test_sessionmaker()
    try:
        yield session
    finally:
        session.close()
        Base.metadata.drop_all(bind=test_db)

@pytest.fixture()
def client():
    """Create a TestClient that uses the test database."""
    app.dependency_overrides[get_session] = get_test_session
    yield TestClient(app)
    del app.dependency_overrides[get_session]

@pytest.fixture(autouse=True, scope="session")
def test_frontmatter():
    """Override the location of frontmatter files."""
    with patch(
        "globoticket.frontmatter.FRONTMATTER_DIRECTORY",
        new=Path(__file__).parent / "product_info",
    ):
        yield

Final tests with frontmatter (test_get_event.py — Module 5)

def test_client_get_event_1(client):
    """Call get_event using the client, retrieving event 1"""
    response = client.get("/event/1")
    assert response.status_code == 200
    assert response.json() == {
        "date": "2024-01-01",
        "id": 1,
        "price": "5.5000000000",
        "artist": "Bob Marley",
        "content": "Love the life you live. Live the life you love.",
        "image": "whalers.jpg",
        "name": "Live at the Rainbow",
    }

def test_client_get_event_404(client):
    """Call get_event using the client, with a non-existing id"""
    response = client.get("/event/100")
    assert response.status_code == 404

6. Complete project structure

Final tree

globoticket/               # Package principal
├── __init__.py
├── api.py                 # Endpoints FastAPI
├── crud.py                # Fonctions d'accès aux données
├── database.py            # Configuration SQLAlchemy
├── frontmatter.py         # Lecture des fichiers frontmatter
├── models.py              # Modèles SQLAlchemy (DBEvent, DBCategory)
└── schemas.py             # Schémas Pydantic (Event)

tests/                     # Tests pytest
├── __init__.py
├── conftest.py            # Fixtures partagées (DB de test, client)
├── product_info/          # Données de test (fichiers .yml)
│   ├── 111222.yml
│   ├── hiphop/
│   │   └── 123456.yml
│   └── reggae/
│       └── 654321.yml
├── test_find_frontmatter_files.py
├── test_get_event.py
└── test_parse_frontmatter.py

product_info/              # Vraies données (fichiers frontmatter)
├── classical/
├── jazz/
└── rock/
    └── 23534643.yml

static/                    # Fichiers HTML statiques
runserver.py               # Script de démarrage
pyproject.toml             # Configuration Poetry
.env                       # Variables d'environnement (non committé)
.pre-commit-config.yaml    # Configuration pre-commit
database.ddl               # Schéma SQL de référence
events.db                  # Base de données SQLite

final pyproject.toml

[tool.poetry]
name = "globoticket"
version = "0.1.0"
description = "Demo project for the Pluralsight course 'Building a REST API with Python 3'"
authors = ["Reindert-Jan Ekker <info@codesensei.nl>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.10"
fastapi = {extras = ["all"], version = "^0.103.1"}
sqlalchemy = "^2.0.20"
python-dotenv = "^1.0.0"
pyyaml = "^6.0.1"

[tool.poetry.group.dev.dependencies]
pytest = "^7.4.1"
pre-commit = "^3.4.0"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

Final globoticket/api.py file

"""
api.py
---

The REST Api for the Globoticket events database.
"""
from pathlib import Path
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from starlette.staticfiles import StaticFiles

from globoticket.crud import get_all_dbevents, get_dbevent
from globoticket.database import SessionLocal
from globoticket.models import DBEvent
from globoticket.schemas import Event

app = FastAPI()

PROJECT_ROOT = Path(__file__).parent.parent

def get_session() -> Session:
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

@app.get("/event/{id}", response_model=Event)
def get_event(id: int, db: Annotated[Session, Depends(get_session)]) -> DBEvent:
    """Retrieve a single event by id. Returns status 404 if event is not found."""
    event = get_dbevent(id, db)
    if event is None:
        raise HTTPException(status_code=404, detail=f"No product with id {id}")
    return event

@app.get("/event/", response_model=list[Event])
def get_all_events(db: Annotated[Session, Depends(get_session)]) -> list[DBEvent]:
    return get_all_dbevents(db)

# This should come AFTER all other endpoints
app.mount("/", StaticFiles(directory=PROJECT_ROOT / "static", html=True))

Final globoticket/frontmatter.py file

import re
from pathlib import Path
from typing import Any

import yaml

FRONTMATTER_DIRECTORY = Path(__file__).parent.parent / "product_info"

def get_frontmatter(product_code: str) -> dict[str, Any]:
    """Read the frontmatter for event with given product_code,
    and return the properties as a dict."""
    frontmatter = find_frontmatter_file(product_code, FRONTMATTER_DIRECTORY).read_text()
    return parse_frontmatter(frontmatter)

def find_frontmatter_file(product_code: str, frontmatter_dir: Path) -> Path:
    """Find a file named `product_code`.yml in frontmatter_dir
    and its subdirectories.

    Raises FileNotFoundError if no file is found for this product_code."""
    matches = list(frontmatter_dir.glob(f"**/{product_code}.yml"))
    if not matches:
        raise FileNotFoundError(
            f"Cannot find yaml file for product {product_code} in {frontmatter_dir}"
        )
    return matches[0]

class InvalidFrontmatterError(Exception):
    pass

FRONTMATTER_REGEX = re.compile(
    r"---\n(?P<yaml>.*?)---\n(?P<content>.*)",
    re.DOTALL,
)

def parse_frontmatter(frontmatter: str) -> dict[str, Any]:
    """Return the contents of frontmatter as a dict.

    Raises an InvalidFrontmatterError if anything is wrong."""

    match = re.match(FRONTMATTER_REGEX, frontmatter)
    if not match:
        raise InvalidFrontmatterError("Invalid file structure")

    try:
        data = yaml.load(match.group("yaml"), Loader=yaml.FullLoader)
        data["content"] = match.group("content") or ""
    except (yaml.YAMLError, TypeError) as e:
        raise InvalidFrontmatterError("Invalid YAML (should be a dict)") from e

    return data

Final globoticket/models.py file

import decimal
from datetime import date

from sqlalchemy import ForeignKey
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    declarative_base,
    mapped_column,
    reconstructor,
    relationship,
)

from globoticket.frontmatter import get_frontmatter

Base: DeclarativeBase = declarative_base()

class DBCategory(Base):
    __tablename__ = "category"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    events: Mapped[list["DBEvent"]] = relationship(back_populates="category")

class DBEvent(Base):
    __tablename__ = "event"

    id: Mapped[int] = mapped_column(primary_key=True)
    product_code: Mapped[str] = mapped_column(unique=True)
    date: Mapped[date]
    price: Mapped[decimal.Decimal]
    category_id: Mapped[int] = mapped_column(ForeignKey("category.id"))
    category: Mapped["DBCategory"] = relationship(back_populates="events")

    @reconstructor
    def _get_frontmatter(self):
        frontmatter = get_frontmatter(self.product_code)
        for k, v in frontmatter.items():
            if not hasattr(self, k):
                setattr(self, k, v)

Search Terms

rest · api · python · java · backend · architecture · full-stack · web · globoticket · tests · data · database · endpoint · function · model · version · configuration · conftest.py · frontmatter · pydantic · schema · testgetevent.py · unit · api.py

More Java, Backend Architecture & Full-Stack courses

View all 30

Full-stack Java Development with Spring Boot 3 and React

A job-portal app — Spring Boot 3 backend, JWT/OAuth2 auth, a React frontend and Docker CI/CD on AWS.

Intermediate

Full-stack Java Development with Play Framework and React

Non-blocking REST APIs and WebSockets with Play and Akka Streams, a React+Redux UI and AWS deployment.

Advanced

Microservices Foundations

What microservices are, why and when to use them, communication patterns and how to adopt them.

Beginner

Asynchronous Programming in Java

By the end of this course, you will be fully equipped to understand the structure of the CompletableFuture API and navigate its many methods.

Intermediate

Building an Application Using Java SE

This course teaches how to build a true end-to-end Java application, using industry standard tools, libraries, and practices. The objective is not to learn a particular framework, but to...

Beginner

Collections in Java SE 21

I am Richard Warburton, Software Developer, Java Champion, author of Pluralsight books and courses in the field of software development and Java.

Intermediate

Interested in this course?

Contact us to book it or get a custom training plan for your team.

Get in Touch Back to Java, Backend Architecture & Full-Stack