Agile, Scalable APIs & Python

The demand for fast, scalable, and easy‑to‑maintain APIs has been growing exponentially. In the same vein, developing asynchronous APIs has become essential for applications that require high concurrency and low latency.
This article presents the construction of a “simple” API focused on managing CrossFit trainings, built as a practical lab to demonstrate the full use of Python, FastAPI, and the SQLAlchemy ORM in an end‑to‑end asynchronous flow, including integration tests with PyTest.
To ensure code quality and endpoint reliability, we incorporated integrated tests using PyTest with support for asynchronous operations. These tests cover everything from validating Pydantic models to verifying the behaviour of the persistence layer, providing rapid feedback during development.
The ultimate goal is to deliver a functional prototype that illustrates best practices in architecture, API documentation, code organisation, and performance optimisation in asynchronous environments. Additionally, it offers a well‑structured, functional example that can serve as a starting point for real projects requiring high performance, simplified maintenance, and comprehensive test coverage—whether for sports training or any other application needing a modern, responsive API.

Functional Requirements

For the lab, the considered functionalities are:

1. A CrossFit athlete belongs to only one category (CrossFit level).
2. Each category can contain multiple athletes.
3. An athlete can train at different training centres.
4. Training centres can host several athletes training simultaneously.

The Technologies

Why FastAPI?

FastAPI was chosen as the primary framework for two fundamental reasons:

• Declarative Model + Automatic Validation – It uses type hints and Pydantic to validate payloads with minimal extra effort.
• Native Async/Await Support – Each route can be declared as async, allowing the server (Uvicorn or Hypercorn) to handle thousands of concurrent connections without blocking the event loop.

These characteristics drastically reduce boilerplate and deliver high performance out‑of‑the‑box.

Asynchronous Persistence with SQLAlchemy

SQLAlchemy provides an asynchronous driver (asyncpg for PostgreSQL).

It combines:

• Full‑featured ORM – Maps Python classes to relational tables.
• Asynchronous Session – CRUD operations can be executed inside Python coroutines, preserving transactional consistency.
• Alembic Integration – Schema migrations are managed to take advantage of the asyncpg driver’s asynchronous implementation.

Partial example of the Domain Model:

from datetime import datetime, timezone

from sqlalchemy import DateTime, ForeignKey, String, func
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class ORMBase(AsyncAttrs, DeclarativeBase):
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=datetime.now(timezone.utc),
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=datetime.now(timezone.utc),
        server_default=func.now(),
        server_onupdate=func.now(),
        onupdate=func.now(),
    )

class Category(ORMBase):
    __tablename__ = "category"
    name: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)

    def __repr__(self):
        return f"<Category(id={self.id}, name={self.name})>"

    athletes: Mapped[list["AthleteCategory"]] = relationship(
        back_populates="categories", cascade="all, delete-orphan"
    )

API Layer with FastAPI

Main directory structure:

├── docs                # Documentation (solution design, ER, UML, OpenAPI)
├── alembic             # Database migration scripts
└── src
    ├── api
    │   ├── controllers   # Feature handlers
    │   ├── main.py       # Entry point
    │   ├── models
    │   │   ├── dto       # Pydantic models (DTO)
    │   │   └── orm       # SQLAlchemy models
    │   ├── services      # Service handlers (e.g., async DB context manager, bus)
    │   ├── setup         # Base setup (API, DB connection)
    │   └── views         # Route definitions
    └── tests
        ├── category      # Test cases for Category
        ├── conftest.py   # Test fixtures
        └── health        # Health‑check test cases

Migrations with Alembic (Asynchronous Mode)

Although Alembic is synchronous by default, it was adapted to work with the asyncpg driver. It is also part of the CI/CD pipeline, where migrations are applied before running tests or deploying to staging/production.

Partial adapted env.py example:

# ...

def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations() -> None:
    """Create an engine and associate a connection with the context."""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()
# ...

Automated Tests with PyTest (Asynchronous)

Database fixtures could be in‑memory, but for integration testing the routines run against the actual application, validating the real database modules.
Partial example of Category test cases:

class TestCategory:
    async def test_category_post_success(
        self, aioclient: AsyncClient, category_in, category_out
    ):
        # Given – set up the stage (category_in)
        # When – perform the action
        response = await aioclient.post(url=self._CATEGORY_URL_PREFIX, json=category_in)
        # Then – verify the outcome
        assert response.status_code == status.HTTP_201_CREATED
        content = response.json()
        assert content["name"] == category_in["name"]
        category_out["id"] = content["id"]
        category_out["name"] = content["name"]
        # ...

The tests use httpx.AsyncClient to run the API in memory, ensuring the whole flow—Pydantic validation, CRUD layer, and asynchronous session—is exercised. Moreover, the CI/CD pipeline includes test‑coverage analysis and code listing.

Conclusion

I’ve shown how to combine the most modern tools in the Python ecosystem to build a fully asynchronous, testable, and production‑ready API. While the use case focuses on CrossFit training management, the presented architecture - FastAPI + SQLAlchemy + Alembic + PyTest - can be reapplied to any domain that demands:

• Real‑time responses
• Horizontal scalability
• Simplified maintenance thanks to strong typing and automatic validation
• Quality assurance through integrated testing

From here, you can extend the solution by adding JWT authentication, WebSockets for live result pushes, integrating a distributed cache (e.g., Valkey or Redis) to lower query latency, or implementing event‑driven components (e.g., Kafka) to expose data for further analysis.

Check out the source code at https://gitlab.com/adrianovieira/workout-api for the API described above.

Let’s keep having fun… ;)

I used the generative AI tools Grammarly and Lumo AI to support me in creating this document in English.