Weird Sheep Labs logo
Contact
export

Fast and furious: async testing with FastAPI and pytest

2024-11-22 | 8 min read
Armand Rego
FastAPI's asynchronous capabilities make it one of the best performing Python web frameworks available, something that has contributed to its meteoric rise in popularity over the past few years. However, because of its use of asynchronous code, testing a FastAPI application can be more complex than a standard synchronous API. In this post, we'll explore how to use pytest and pytest-asyncio to write effective async tests for your application.

Demo application

The code repository that forms the basis of this post is public and can be found at https://github.com/Weird-Sheep-Labs/fastapi-pytest-asyncio-demo. The demo application is a simple CRUD API to interact with a Song database model. The structure of the repo follows the src layout, with tests in a separate directory:

1 
└── 📁src
2 
    └── 📁filters
3 
        └── __init__.py
4 
        └── songs.py
5 
    └── 📁models
6 
        └── __init__.py
7 
        └── base.py
8 
        └── songs.py
9 
    └── 📁routers
10 
        └── __init__.py
11 
        └── songs.py
12 
    └── db.py
13 
    └── main.py
14 
└── 📁tests
15 
    └── 📁fixtures
16 
        └── spotify_top_100_20241119.csv
17 
    └── 📁routers
18 
        └── test_songs.py
19 
    └── __init__.py
20 
    └── conftest.py
21 
    └── factories.py

We'll first highlight some of the most relevant sections from the application code first, before going over the main meat of the topic, the test setup, in more detail.

Application code

In order to take advantage of FastAPI's asynchronous capabilities, we need to interact with I/O and make network requests using async libraries - a typical example of this would be the database driver. We connect to the Postgres DB using asyncpg, a highly performant, async DB driver for Python and Postgres, by specifying the driver in the database connection string in a .env at the root of project:

DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/fastapi-pytest-asyncio-demo

We then configure the DB engine and session in src/db.py:

1 
import os
2
3 
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
4 
from sqlmodel import SQLModel
5 
from sqlmodel.ext.asyncio.session import AsyncSession
6
7 
DATABASE_URL = os.environ["DATABASE_URL"]
8
9 
engine = create_async_engine(DATABASE_URL, echo=True, pool_pre_ping=True)
10 
async_session = async_sessionmaker(
11 
    engine, class_=AsyncSession, autocommit=False, autoflush=False
12 
)
13
14
15 
async def init_db():
16 
    async with engine.begin() as conn:
17 
        await conn.run_sync(SQLModel.metadata.create_all)
18
19
20 
async def get_session():
21 
    async with async_session() as session:
22 
        yield session

We use methods from SQLAlchemy's asyncio extension to create the DB engine and session, very similar to how this would be done in a sync context. Note that not only is get_session() now defined as an async method, but the context manager that yields the session is also async, which allows any failed transactions to be implicitly handled in the __aexit__() context manager method.

SQLModel

As we are using SQLModel for the ORM, we specify class_=AsyncSession when calling the session maker otherwise the returned session class would be an instance of the default sqlalchemy.ext.asyncio.AsyncSession. SQLModel brings significant benefits over SQLAlchemy when used within a FastAPI application, largely due to the fact a sqlmodel.BaseModel is a subclass of both sqlalchemy.BaseModel and pydantic.BaseModel! This avoids having to define models twice - once for the DB model itself and once as a request payload validation model.

Test setup

As the path operators (i.e router methods) of our application are async functions, we need to await them in our tests; pytest-asyncio is what allows us to do this.

Configuration

In order to achieve maximum test isolation, we choose to scope the fixtures, tests and event loop at the function level by specifying asyncio_default_fixture_loop_scope = "function" in the pyproject.toml at the root of the project. The demo repository includes a class_scope branch with the test setup configured to be scoped at the class level. This can result in a faster-running test suite as the test DB does not have to be setup and torn down after each test. Additionally, it can also be useful if individual tests are necessarily coupled via shared dependencies, however, this coupling can also lead to hard-to-debug errors within the test code itself.

Fixtures

The tests/conftest.py file contains the async engine, session and client fixtures that are available to use in the tests:

1 
import os
2
3 
import pytest_asyncio
4 
from httpx import ASGITransport, AsyncClient
5 
from sqlalchemy.ext.asyncio import (
6 
    AsyncSession,
7 
    async_sessionmaker,
8 
    create_async_engine,
9 
)
10 
from sqlalchemy.pool import NullPool
11 
from sqlmodel import SQLModel
12
13 
from db import get_session
14 
from main import app
15
16 
async_engine = create_async_engine(
17 
    url=os.environ["DATABASE_URL"],
18 
    echo=False,
19 
    poolclass=NullPool,
20 
)
21
22
23 
# Drop all tables after each test
24 
@pytest_asyncio.fixture(scope="function")
25 
async def async_db_engine():
26 
    async with async_engine.begin() as conn:
27 
        await conn.run_sync(SQLModel.metadata.create_all)
28
29 
    yield async_engine
30
31 
    async with async_engine.begin() as conn:
32 
        await conn.run_sync(SQLModel.metadata.drop_all)
33
34
35 
@pytest_asyncio.fixture(scope="function")
36 
async def async_db(async_db_engine):
37 
    async_session = async_sessionmaker(
38 
        expire_on_commit=False,
39 
        autocommit=False,
40 
        autoflush=False,
41 
        bind=async_db_engine,
42 
        class_=AsyncSession,
43 
    )
44
45 
    async with async_session() as session:
46 
        await session.begin()
47
48 
        yield session
49
50 
        await session.rollback()
51
52
53 
@pytest_asyncio.fixture(scope="function", autouse=True)
54 
async def async_client(async_db):
55 
    def override_get_db():
56 
        yield async_db
57
58 
    app.dependency_overrides[get_session] = override_get_db
59 
    return AsyncClient(transport=ASGITransport(app=app), base_url="http://localhost")

We define an async engine similar to that in the application code itself, however, due to the fact that we have scoped the test setup at the function level, we disable connection pooling with poolclass=NullPool. The engine is wrapped in an async fixture which creates the DB tables and drops them before and after each test function, allowing for maximum test isolation.

The DB session fixture is defined similarly; beginning a session before each test, and rolling it back after. Using FastAPI's dependency injection mechanism, we then provide the session to an httpx async client, overriding the get_session() method from the application. We can then use this client to make HTTP calls to the API endpoints and make assertions about their responses:

1 
@pytest.mark.asyncio
2 
    async def test_list_songs(self, async_client):
3 
        """
4 
        Tests that the songs are listed correctly.
5 
        """
6 
        resp = await async_client.get("/v1/songs")
7 
        assert resp.status_code == 200
8 
        assert resp.json()["limit"] == 50
9 
        assert resp.json()["offset"] == 0
10 
        assert resp.json()["total"] == 100

Conclusion

Testing is a critical component of building systems that are reliable and resistant to the inadvertent introduction of regressions. In this post, we have shown how a FastAPI application can be set up with asynchronous tests using pytest and pytest-asyncio, allowing for faster iteration and development velocity.

At Weird Sheep Labs, we have built APIs for multiple clients that use the technologies mentioned in this post. If this sounds like something your company could benefit from, don't hesitate to get in touch!

Work
Company
Connect
© Weird Sheep Labs Ltd 2025
Weird Sheep Labs Ltd is a company registered in England & Wales (Company No. 15160367)
85 Great Portland St, London, W1W 7LT