Developing and Testing an Asynchronous API with FastAPI and Pytest

Posted by Michael Herman Michael Herman Last updated December 1st, 2022

This tutorial looks at how to develop and test an asynchronous API with FastAPI, Postgres, pytest and Docker using Test-driven Development (TDD). We'll also use the Databases package for interacting with Postgres asynchronously.

Dependencies:

  1. FastAPI v0.88.0
  2. Docker v20.10.21
  3. Python v3.11.0
  4. pytest v7.2.0
  5. Databases v0.6.2

Contents

Objectives

By the end of this tutorial you should be able to:

  1. Develop an asynchronous RESTful API with Python and FastAPI
  2. Practice Test-driven Development
  3. Test a FastAPI app with pytest
  4. Interact with a Postgres database asynchronously
  5. Containerize FastAPI and Postgres inside a Docker container
  6. Parameterize test functions and mock functionality in tests with pytest
  7. Document a RESTful API with Swagger/OpenAPI

FastAPI

FastAPI is a modern, high-performance, batteries-included Python web framework that's perfect for building RESTful APIs. It can handle both synchronous and asynchronous requests and has built-in support for data validation, JSON serialization, authentication and authorization, and OpenAPI (version 3.0.2 as of writing) documentation.

Highlights:

  1. Heavily inspired by Flask, it has a lightweight microframework feel with support for Flask-like route decorators.
  2. It takes advantage of Python type hints for parameter declaration which enables data validation (via Pydantic) and OpenAPI/Swagger documentation.
  3. Built on top of Starlette, it supports the development of asynchronous APIs.
  4. It's fast. Since async is much more efficient than the traditional synchronous threading model, it can compete with Node and Go with regards to performance.

Review the Features guide from the official docs for more info. It's also encouraged to review Alternatives, Inspiration, and Comparisons, which details how FastAPI compares to other web frameworks and technologies, for context.

Project Setup

Start by creating a folder to hold your project called "fastapi-crud". Then, add a docker-compose.yml file and a "src" folder to the project root. Within the "src" folder, add a Dockerfile, requirements.txt file, and an "app" folder. Finally, add the following files to the "app" folder: __init__.py and main.py.

The following command will create the project structure:

$ mkdir fastapi-crud && \
    cd fastapi-crud && \
    touch docker-compose.yml && \
    mkdir src && \
    cd src && \
    touch Dockerfile && \
    touch requirements.txt && \
    mkdir app && \
    cd app && \
    touch __init__.py && \
    touch main.py

You should now have:

fastapi-crud
    ├── docker-compose.yml
    └── src
        ├── Dockerfile
        ├── app
        │   ├── __init__.py
        │   └── main.py
        └── requirements.txt

Unlike Django or Flask, FastAPI does not have a built-in development server. So, we'll use Uvicorn, an ASGI server, to serve up FastAPI.

New to ASGI? Read through the excellent Introduction to ASGI: Emergence of an Async Python Web Ecosystem article.

Add FastAPI and Uvicorn to the requirements file:

fastapi==0.88.0
uvicorn==0.20.0

The fact that FastAPI does not come with a development server is both a positive and a negative in my opinion. On the one hand, it does take a bit more to serve up the app in development mode. On the other, this helps to conceptually separate the web framework from the web server, which is often a source of confusion for beginners when one moves from development to production with a web framework that does have a built-in development server (like Django or Flask).

Then, within main.py, create a new instance of FastAPI and set up a sanity check route:

from fastapi import FastAPI

app = FastAPI()


@app.get("/ping")
def pong():
    return {"ping": "pong!"}

Install Docker, if you don't already have it, and then update the Dockerfile in the "src" directory:

# pull official base image
FROM python:3.11.0-alpine

# set work directory
WORKDIR /usr/src/app

# set environment variables
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

# copy requirements file
COPY ./requirements.txt /usr/src/app/requirements.txt

# install dependencies
RUN set -eux \
    && apk add --no-cache --virtual .build-deps build-base \
         openssl-dev libffi-dev gcc musl-dev python3-dev \
    && pip install --upgrade pip setuptools wheel \
    && pip install -r /usr/src/app/requirements.txt \
    && rm -rf /root/.cache/pip

# copy project
COPY . /usr/src/app/

So, we started with an Alpine-based Docker image for Python 3.11.0. We then set a working directory along with two environment variables:

  1. PYTHONDONTWRITEBYTECODE: Prevents Python from writing pyc files to disc (equivalent to python -B option)
  2. PYTHONUNBUFFERED: Prevents Python from buffering stdout and stderr (equivalent to python -u option)

Finally, we copied over the requirements.txt file, installed some system-level dependencies, updated Pip, installed the requirements, and copied over the FastAPI app itself.

Review Docker Best Practices for Python Developers for more on structuring Dockerfiles as well as some best practices for configuring Docker for Python-based development.

Next, add the following to the docker-compose.yml file in the project root:

version: '3.8'

services:
  web:
    build: ./src
    command: uvicorn app.main:app --reload --workers 1 --host 0.0.0.0 --port 8000
    volumes:
      - ./src/:/usr/src/app/
    ports:
      - 8002:8000

So, when the container spins up, Uvicorn will run with the following settings:

  1. --reload enables auto-reload so the server will restart after changes are made to the code base.
  2. --workers 1 provides a single worker process.
  3. --host 0.0.0.0 defines the address to host the server on.
  4. --port 8000 defines the port to host the server on.

app.main:app tells Uvicorn where it can find the FastAPI ASGI application -- i.e., "within the 'app' module, you'll find the ASGI app, app = FastAPI(), in the 'main.py' file.

For more on the Docker Compose file config, review the Compose file reference.

Build the image and spin up the container:

$ docker-compose up -d --build

Navigate to http://localhost:8002/ping. You should see:

{
  "ping": "pong!"
}

You'll also be able to view the interactive API documentation, powered by Swagger UI, at http://localhost:8002/docs:

swagger ui

Test Setup

Create a "tests" folder in "src" and then add an __init__.py file to "tests" along with a test_main.py file:

from starlette.testclient import TestClient

from app.main import app

client = TestClient(app)


def test_ping():
    response = client.get("/ping")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong!"}

Here, we imported Starlette's TestClient, which uses the httpx library to make requests against the FastAPI app.

Add pytest and httpx to requirements.txt:

fastapi==0.88.0
uvicorn==0.20.0

# dev
pytest==7.2.0
httpx==0.23.1

Update the image and then run the tests:

$ docker-compose up -d --build
$ docker-compose exec web pytest .

You should see:

=============================== test session starts ===============================
platform linux -- Python 3.11.0, pytest-7.2.0, pluggy-1.0.0
rootdir: /usr/src/app
plugins: anyio-3.6.2
collected 1 item

tests/test_main.py .                                                        [100%]

================================ 1 passed in 0.31s ================================

Before moving on, add a test_app pytest fixture to a new file called src/tests/conftest.py:

import pytest
from starlette.testclient import TestClient

from app.main import app


@pytest.fixture(scope="module")
def test_app():
    client = TestClient(app)
    yield client  # testing happens here

Update the test file as well so that it uses the fixture:

def test_ping(test_app):
    response = test_app.get("/ping")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong!"}

Your project structure should now look like this:

fastapi-crud
    ├── docker-compose.yml
    └── src
        ├── Dockerfile
        ├── app
        │   ├── __init__.py
        │   └── main.py
        ├── requirements.txt
        └── tests
            ├── __init__.py
            ├── conftest.py
            └── test_main.py

Async Handlers

Let's convert the synchronous handler over to an asynchronous one.

Rather than having to go through the trouble of spinning up a task queue (like Celery or RQ) or utilizing threads, FastAPI makes it easy to deliver routes asynchronously. As long as you don't have any blocking I/O calls in the handler, you can simply declare the handler as asynchronous by adding the async keyword like so:

@app.get("/ping")
async def pong():
    # some async operation could happen here
    # example: `notes = await get_all_notes()`
    return {"ping": "pong!"}

That's it. Update the handler in your code, and then make sure the tests still pass:

=============================== test session starts ===============================
platform linux -- Python 3.11.0, pytest-7.2.0, pluggy-1.0.0
rootdir: /usr/src/app
plugins: anyio-3.6.2
collected 1 item

tests/test_main.py .                                                        [100%]

================================ 1 passed in 0.06s ================================

Review the Concurrency and async / await guide for a technical deep dive into async.

Routes

Next, let's set up the basic CRUD routes, following RESTful best practices:

Endpoint HTTP Method CRUD Method Result
/notes/ GET READ get all notes
/notes/:id/ GET READ get a single note
/notes/ POST CREATE add a note
/notes/:id/ PUT UPDATE update a note
/notes/:id/ DELETE DELETE delete a note

For each route, we'll:

  1. write a test
  2. run the test, to ensure it fails (red)
  3. write just enough code to get the test to pass (green)
  4. refactor (if necessary)

Before diving in, let's add some structure to better organize the CRUD routes with FastAPI's APIRouter.

You can break up and modularize larger projects as well as apply versioning to your API with the APIRouter. If you're familiar with Flask, it is equivalent to a Blueprint.

First, add a new folder called "api" to the "app" folder. Add an __init__.py file to the newly created folder.

Now we can move the /ping route to a new file called src/app/api/ping.py:

from fastapi import APIRouter

router = APIRouter()


@router.get("/ping")
async def pong():
    # some async operation could happen here
    # example: `notes = await get_all_notes()`
    return {"ping": "pong!"}

Then, update main.py like so to remove the old route and wire the router up to our main app:

from fastapi import FastAPI

from app.api import ping

app = FastAPI()


app.include_router(ping.router)

Rename test_main.py to test_ping.py.

Make sure http://localhost:8002/ping and http://localhost:8002/docs still work. Also, be sure the tests still pass before moving on.

fastapi-crud
    ├── docker-compose.yml
    └── src
        ├── Dockerfile
        ├── app
        │   ├── __init__.py
        │   ├── api
        │   │   ├── __init__.py
        │   │   └── ping.py
        │   └── main.py
        ├── requirements.txt
        └── tests
            ├── __init__.py
            ├── conftest.py
            └── test_ping.py

Postgres Setup

To configure Postgres, we'll need to add a new service to the docker-compose.yml file, add the appropriate environment variables, and install asyncpg.

First, add a new service called db to docker-compose.yml:

version: '3.8'

services:
  web:
    build: ./src
    command: |
      bash -c 'while !</dev/tcp/db/5432; do sleep 1; done; uvicorn app.main:app --reload --workers 1 --host 0.0.0.0 --port 8000'
    volumes:
      - ./src/:/usr/src/app/
    ports:
      - 8002:8000
    environment:
      - DATABASE_URL=postgresql://hello_fastapi:hello_fastapi@db/hello_fastapi_dev
  db:
    image: postgres:15.1-alpine
    volumes:
      - postgres_data:/var/lib/postgresql/data/
    expose:
      - 5432
    environment:
      - POSTGRES_USER=hello_fastapi
      - POSTGRES_PASSWORD=hello_fastapi
      - POSTGRES_DB=hello_fastapi_dev

volumes:
  postgres_data:

To persist the data beyond the life of the container we configured a volume. This config will bind postgres_data to the "/var/lib/postgresql/data/" directory in the container.

We also added an environment key to define a name for the default database and set a username and password.

Review the "Environment Variables" section of the Postgres Docker Hub page for more info.

Update the Dockerfile to install the appropriate packages required for asyncpg:

# pull official base image
FROM python:3.11.0-alpine

# set work directory
WORKDIR /usr/src/app

# set environment variables
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

# copy requirements file
COPY ./requirements.txt /usr/src/app/requirements.txt

# install dependencies
RUN set -eux \
    && apk add --no-cache --virtual .build-deps build-base \
         openssl-dev libffi-dev gcc musl-dev python3-dev \
        postgresql-dev bash \
    && pip install --upgrade pip setuptools wheel \
    && pip install -r /usr/src/app/requirements.txt \
    && rm -rf /root/.cache/pip

# copy project
COPY . /usr/src/app/

Add asyncpg to src/requirements.txt:

asyncpg==0.27.0
fastapi==0.88.0
uvicorn==0.20.0

# dev
pytest==7.2.0
httpx==0.23.1

Next, add a db.py file to "src/app":

import os

from databases import Database
from sqlalchemy import create_engine, MetaData


DATABASE_URL = os.getenv("DATABASE_URL")

# SQLAlchemy
engine = create_engine(DATABASE_URL)
metadata = MetaData()

# databases query builder
database = Database(DATABASE_URL)

Here, using the database URI and credentials that we just configured in the Docker Compose file, we created a SQLAlchemy engine (used for communicating with the database) along with a Metadata instance (used for creating the database schema). We also created a new Database instance from Databases.

Databases is an async SQL query builder that works on top of the SQLAlchemy Core expression language. It supports the following methods:

  1. database.fetch_all(query)
  2. database.fetch_one(query)
  3. database.iterate(query)
  4. database.execute(query)
  5. database.execute_many(query)

Review the Async SQL (Relational) Databases guide and the Starlette Database docs for more details on working with databases asynchronously.

Update the requirements:

asyncpg==0.27.0
databases[postgresql]==0.6.2
fastapi==0.88.0
psycopg2-binary==2.9.5
SQLAlchemy==1.4.41
uvicorn==0.20.0

# dev
pytest==7.2.0
httpx==0.23.1

We're installing Psycopg since we will be using create_all, which is a synchronous SQLAlchemy function.

Models

SQLAlchemy Model

Add a notes model to src/app/db.py:

import os

from sqlalchemy import (
    Column,
    DateTime,
    Integer,
    MetaData,
    String,
    Table,
    create_engine
)
from sqlalchemy.sql import func

from databases import Database

DATABASE_URL = os.getenv("DATABASE_URL")

# SQLAlchemy
engine = create_engine(DATABASE_URL)
metadata = MetaData()
notes = Table(
    "notes",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("title", String(50)),
    Column("description", String(50)),
    Column("created_date", DateTime, default=func.now(), nullable=False),
)

# databases query builder
database = Database(DATABASE_URL)

Wire up the database and the model in main.py and add startup and shutdown event handlers for connecting to and disconnecting from the database:

from fastapi import FastAPI

from app.api import ping
from app.db import engine, database, metadata

metadata.create_all(engine)

app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


app.include_router(ping.router)

Build the new image and spin up the two containers:

$ docker-compose up -d --build

Ensure the notes table was created:

$ docker-compose exec db psql --username=hello_fastapi --dbname=hello_fastapi_dev

psql (15.1)
Type "help" for help.

hello_fastapi_dev=# \l
                                            List of databases
       Name        |     Owner     | Encoding |  Collate   |   Ctype    |        Access privileges
-------------------+---------------+----------+------------+------------+---------------------------------
 hello_fastapi_dev | hello_fastapi | UTF8     | en_US.utf8 | en_US.utf8 |
 postgres          | hello_fastapi | UTF8     | en_US.utf8 | en_US.utf8 |
 template0         | hello_fastapi | UTF8     | en_US.utf8 | en_US.utf8 | =c/hello_fastapi               +
                   |               |          |            |            | hello_fastapi=CTc/hello_fastapi
 template1         | hello_fastapi | UTF8     | en_US.utf8 | en_US.utf8 | =c/hello_fastapi               +
                   |               |          |            |            | hello_fastapi=CTc/hello_fastapi
(4 rows)

hello_fastapi_dev=# \c hello_fastapi_dev
You are now connected to database "hello_fastapi_dev" as user "hello_fastapi".

hello_fastapi_dev=# \dt
           List of relations
 Schema | Name  | Type  |     Owner
--------+-------+-------+---------------
 public | notes | table | hello_fastapi
(1 row)

hello_fastapi_dev=# \q

Pydantic Model

First time using Pydantic? Review the Overview guide from the official docs.

Create a NoteSchema Pydantic model with two required fields, title and description, in a new file called models.py in "src/app/api":

from pydantic import BaseModel


class NoteSchema(BaseModel):
    title: str
    description: str

NoteSchema will be used for validating the payloads for creating and updating notes.

POST Route

Let's break from the normal TDD flow for this first route in order to establish the coding pattern that we'll use for the remaining routes.

Code

Create a new file called notes.py in the "src/app/api" folder:

from fastapi import APIRouter, HTTPException

from app.api import crud
from app.api.models import NoteDB, NoteSchema

router = APIRouter()


@router.post("/", response_model=NoteDB, status_code=201)
async def create_note(payload: NoteSchema):
    note_id = await crud.post(payload)

    response_object = {
        "id": note_id,
        "title": payload.title,
        "description": payload.description,
    }
    return response_object

Here, we defined a handler that expects a payload, payload: NoteSchema, with a title and a description.

Essentially, when the route is hit with a POST request, FastAPI will read the body of the request and validate the data:

  • If valid, the data will be available in the payload parameter. FastAPI also generates JSON Schema definitions that are then used to automatically generate the OpenAPI schema and the API documentation.
  • If invalid, an error is immediately returned.

Review the Request Body docs for more info.

It's worth noting that we used the async declaration here since the database communication will be asynchronous. In other words, there are no blocking I/O operations in the handler.

Next, create a new file called crud.py in the "src/app/api" folder:

from app.api.models import NoteSchema
from app.db import notes, database


async def post(payload: NoteSchema):
    query = notes.insert().values(title=payload.title, description=payload.description)
    return await database.execute(query=query)

We added a utility function called post for creating new notes that takes a payload object and then:

  1. Creates a SQLAlchemy insert object expression query
  2. Executes the query and returns the generated ID

Next, we need to define a new Pydantic model for use as the response_model:

@router.post("/", response_model=NoteDB, status_code=201)

Update models.py like so:

from pydantic import BaseModel


class NoteSchema(BaseModel):
    title: str
    description: str


class NoteDB(NoteSchema):
    id: int

The NoteDB model inherits from the NoteSchema model, adding an id field.

Wire up the new router in main.py:

from fastapi import FastAPI

from app.api import notes, ping
from app.db import database, engine, metadata

metadata.create_all(engine)

app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


app.include_router(ping.router)
app.include_router(notes.router, prefix="/notes", tags=["notes"])

Take note of the prefix URL along with the "notes" tag, which will be applied to the OpenAPI schema (for grouping operations).

Test it out with curl or HTTPie:

$ http --json POST http://localhost:8002/notes/ title=foo description=bar

You should see:

HTTP/1.1 201 Created
content-length: 42
content-type: application/json
date: Wed, 23 Nov 2022 18:14:51 GMT
server: uvicorn

{
    "description": "bar",
    "id": 1,
    "title": "foo"
}

You can also interact with the endpoint at http://localhost:8002/docs.

Test

Add the following test to a new test file called src/tests/test_notes.py:

import json

import pytest

from app.api import crud


def test_create_note(test_app, monkeypatch):
    test_request_payload = {"title": "something", "description": "something else"}
    test_response_payload = {"id": 1, "title": "something", "description": "something else"}

    async def mock_post(payload):
        return 1

    monkeypatch.setattr(crud, "post", mock_post)

    response = test_app.post("/notes/", content=json.dumps(test_request_payload),)

    assert response.status_code == 201
    assert response.json() == test_response_payload


def test_create_note_invalid_json(test_app):
    response = test_app.post("/notes/", content=json.dumps({"title": "something"}))
    assert response.status_code == 422

This test uses the pytest monkeypatch fixture to mock out the crud.post function. We then asserted that the endpoint responds with the expected status codes and response body.

$ docker-compose exec web pytest .

=============================== test session starts ===============================
platform linux -- Python 3.11.0, pytest-7.2.0, pluggy-1.0.0
rootdir: /usr/src/app
plugins: anyio-3.6.2
collected 3 items

tests/test_notes.py ..                                                      [ 66%]
tests/test_ping.py .                                                        [100%]

================================ 3 passed in 0.08s ================================

With that, we can configure the remaining CRUD routes using Test-driven Development.

fastapi-crud
    ├── docker-compose.yml
    └── src
        ├── Dockerfile
        ├── app
        │   ├── __init__.py
        │   ├── api
        │   │   ├── __init__.py
        │   │   ├── crud.py
        │   │   ├── models.py
        │   │   ├── notes.py
        │   │   └── ping.py
        │   ├── db.py
        │   └── main.py
        ├── requirements.txt
        └── tests
            ├── __init__.py
            ├── conftest.py
            ├── test_notes.py
            └── test_ping.py

GET Routes

Test

Add the following tests to src/tests/test_notes.py:

def test_read_note(test_app, monkeypatch):
    test_data = {"id": 1, "title": "something", "description": "something else"}

    async def mock_get(id):
        return test_data

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.get("/notes/1")
    assert response.status_code == 200
    assert response.json() == test_data


def test_read_note_incorrect_id(test_app, monkeypatch):
    async def mock_get(id):
        return None

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.get("/notes/999")
    assert response.status_code == 404
    assert response.json()["detail"] == "Note not found"

They should fail:

=============================== test session starts ===============================
platform linux -- Python 3.11.0, pytest-7.2.0, pluggy-1.0.0
rootdir: /usr/src/app
plugins: anyio-3.6.2
collected 5 items

tests/test_notes.py ..FF                                                    [ 80%]
tests/test_ping.py .                                                        [100%]

==================================== FAILURES =====================================
_________________________________ test_read_note __________________________________

test_app = <starlette.testclient.TestClient object at 0x7f29072dc390>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f290734d5d0>

    def test_read_note(test_app, monkeypatch):
        test_data = {"id": 1, "title": "something", "description": "something else"}

        async def mock_get(id):
            return test_data

>       monkeypatch.setattr(crud, "get", mock_get)
E       AttributeError: <module 'app.api.crud' from '/usr/src/app/app/api/crud.py'> has no attribute 'get'

tests/test_notes.py:35: AttributeError
___________________________ test_read_note_incorrect_id ___________________________

test_app = <starlette.testclient.TestClient object at 0x7f29072dc390>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f290729ead0>

    def test_read_note_incorrect_id(test_app, monkeypatch):
        async def mock_get(id):
            return None

>       monkeypatch.setattr(crud, "get", mock_get)
E       AttributeError: <module 'app.api.crud' from '/usr/src/app/app/api/crud.py'> has no attribute 'get'

tests/test_notes.py:46: AttributeError
============================= short test summary info =============================
FAILED tests/test_notes.py::test_read_note - AttributeError: <module 'app.api.crud' from '/usr/src/app/app/api/crud.py'> ha...
FAILED tests/test_notes.py::test_read_note_incorrect_id - AttributeError: <module 'app.api.crud' from '/usr/src/app/app/api/crud.py'> ha...
=========================== 2 failed, 3 passed in 0.11s ===========================

Code

Add the handler function to src/app/api/notes.py:

@router.get("/{id}/", response_model=NoteDB)
async def read_note(id: int):
    note = await crud.get(id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    return note

Here, instead of taking a payload, the handler requires an id, an integer, which will come from the path -- e.g., /notes/5/.

Add the get utility function to crud.py:

async def get(id: int):
    query = notes.select().where(id == notes.c.id)
    return await database.fetch_one(query=query)

Before moving on, ensure the tests pass and manually test the new endpoint in the browser, with curl or HTTPie, and/or via the API documentation.

Test

Next, add a test for reading all notes:

def test_read_all_notes(test_app, monkeypatch):
    test_data = [
        {"title": "something", "description": "something else", "id": 1},
        {"title": "someone", "description": "someone else", "id": 2},
    ]

    async def mock_get_all():
        return test_data

    monkeypatch.setattr(crud, "get_all", mock_get_all)

    response = test_app.get("/notes/")
    assert response.status_code == 200
    assert response.json() == test_data

Again, make sure the test fails.

Code

Add the handler function to src/app/api/notes.py:

@router.get("/", response_model=List[NoteDB])
async def read_all_notes():
    return await crud.get_all()

Import List from Python's typing module:

from typing import List

The response_model is a List with a NoteDB subtype.

Add the CRUD util:

async def get_all():
    query = notes.select()
    return await database.fetch_all(query=query)

Make sure the automated tests pass. Manually test this endpoint as well.

PUT Route

Test

def test_update_note(test_app, monkeypatch):
    test_update_data = {"title": "someone", "description": "someone else", "id": 1}

    async def mock_get(id):
        return True

    monkeypatch.setattr(crud, "get", mock_get)

    async def mock_put(id, payload):
        return 1

    monkeypatch.setattr(crud, "put", mock_put)

    response = test_app.put("/notes/1/", content=json.dumps(test_update_data))
    assert response.status_code == 200
    assert response.json() == test_update_data


@pytest.mark.parametrize(
    "id, payload, status_code",
    [
        [1, {}, 422],
        [1, {"description": "bar"}, 422],
        [999, {"title": "foo", "description": "bar"}, 404],
    ],
)
def test_update_note_invalid(test_app, monkeypatch, id, payload, status_code):
    async def mock_get(id):
        return None

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.put(f"/notes/{id}/", content=json.dumps(payload),)
    assert response.status_code == status_code

This test uses the pytest parametrize decorator to parametrize the arguments for the test_update_note_invalid function.

Code

Handler:

@router.put("/{id}/", response_model=NoteDB)
async def update_note(id: int, payload: NoteSchema):
    note = await crud.get(id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    note_id = await crud.put(id, payload)

    response_object = {
        "id": note_id,
        "title": payload.title,
        "description": payload.description,
    }
    return response_object

Util:

async def put(id: int, payload: NoteSchema):
    query = (
        notes
        .update()
        .where(id == notes.c.id)
        .values(title=payload.title, description=payload.description)
        .returning(notes.c.id)
    )
    return await database.execute(query=query)

DELETE Route

Test

def test_remove_note(test_app, monkeypatch):
    test_data = {"title": "something", "description": "something else", "id": 1}

    async def mock_get(id):
        return test_data

    monkeypatch.setattr(crud, "get", mock_get)

    async def mock_delete(id):
        return id

    monkeypatch.setattr(crud, "delete", mock_delete)

    response = test_app.delete("/notes/1/")
    assert response.status_code == 200
    assert response.json() == test_data


def test_remove_note_incorrect_id(test_app, monkeypatch):
    async def mock_get(id):
        return None

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.delete("/notes/999/")
    assert response.status_code == 404
    assert response.json()["detail"] == "Note not found"

Code

Handler:

@router.delete("/{id}/", response_model=NoteDB)
async def delete_note(id: int):
    note = await crud.get(id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    await crud.delete(id)

    return note

Util:

async def delete(id: int):
    query = notes.delete().where(id == notes.c.id)
    return await database.execute(query=query)

Make sure all tests pass:

=============================== test session starts ===============================
platform linux -- Python 3.11.0, pytest-7.2.0, pluggy-1.0.0
rootdir: /usr/src/app
plugins: anyio-3.6.2
collected 12 items

tests/test_notes.py ...........                                             [ 91%]
tests/test_ping.py .                                                        [100%]

=============================== 12 passed in 0.13s ================================

Additional Validation

Let's add some additional validation to the routes, checking that:

  1. The id is greater than 0 for reading a single note, updating a note, and deleting a note
  2. The title and description fields from the request payloads must have lengths >= 3 and <= 50 for adding and updating a note

GET

Update the test_read_note_incorrect_id test:

def test_read_note_incorrect_id(test_app, monkeypatch):
    async def mock_get(id):
        return None

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.get("/notes/999")
    assert response.status_code == 404
    assert response.json()["detail"] == "Note not found"

    response = test_app.get("/notes/0")
    assert response.status_code == 422

The test should fail:

>       assert response.status_code == 422
E       assert 404 == 422
E        +  where 404 = <Response [404]>.status_code

Update the handler:

@router.get("/{id}/", response_model=NoteDB)
async def read_note(id: int = Path(..., gt=0),):
    note = await crud.get(id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    return note

Make sure to import Path:

from fastapi import APIRouter, HTTPException, Path

So, we added the following metadata to the parameter with Path:

  1. ... - the value is required (Ellipsis)
  2. gt - the value must be greater than 0

The tests should pass. Try out the API documentation as well:

swagger ui

POST

Update the test_create_note_invalid_json test:

def test_create_note_invalid_json(test_app):
    response = test_app.post("/notes/", content=json.dumps({"title": "something"}))
    assert response.status_code == 422

    response = test_app.post("/notes/", content=json.dumps({"title": "1", "description": "2"}))
    assert response.status_code == 422

To get the test to pass, update the NoteSchema model like so:

class NoteSchema(BaseModel):
    title: str = Field(..., min_length=3, max_length=50)
    description: str = Field(..., min_length=3, max_length=50)

Here, we added additional validation to the Pydantic model with Field. It works just like Path.

Add the import:

from pydantic import BaseModel, Field

PUT

Add three more scenarios to test_update_note_invalid:

@pytest.mark.parametrize(
    "id, payload, status_code",
    [
        [1, {}, 422],
        [1, {"description": "bar"}, 422],
        [999, {"title": "foo", "description": "bar"}, 404],
        [1, {"title": "1", "description": "bar"}, 422],
        [1, {"title": "foo", "description": "1"}, 422],
        [0, {"title": "foo", "description": "bar"}, 422],
    ],
)
def test_update_note_invalid(test_app, monkeypatch, id, payload, status_code):
    async def mock_get(id):
        return None

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.put(f"/notes/{id}/", content=json.dumps(payload),)
    assert response.status_code == status_code

Handler:

@router.put("/{id}/", response_model=NoteDB)
async def update_note(payload: NoteSchema, id: int = Path(..., gt=0),):
    note = await crud.get(id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    note_id = await crud.put(id, payload)

    response_object = {
        "id": note_id,
        "title": payload.title,
        "description": payload.description,
    }
    return response_object

DELETE

Test:

def test_remove_note_incorrect_id(test_app, monkeypatch):
    async def mock_get(id):
        return None

    monkeypatch.setattr(crud, "get", mock_get)

    response = test_app.delete("/notes/999/")
    assert response.status_code == 404
    assert response.json()["detail"] == "Note not found"

    response = test_app.delete("/notes/0/")
    assert response.status_code == 422

Handler:

@router.delete("/{id}/", response_model=NoteDB)
async def delete_note(id: int = Path(..., gt=0)):
    note = await crud.get(id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    await crud.delete(id)

    return note

The tests should pass:

=============================== test session starts ===============================
platform linux -- Python 3.11.0, pytest-7.2.0, pluggy-1.0.0
rootdir: /usr/src/app
plugins: anyio-3.6.2
collected 15 items

tests/test_notes.py ..............                                          [ 93%]
tests/test_ping.py .                                                        [100%]

=============================== 15 passed in 0.14s ================================

Synchronous Example

We built a synchronous flavor of this API for so you can compare the two models. You can grab the code from the fastapi-crud-sync repo. Try conducting some performance tests against both versions on your own with ApacheBench.

Conclusion

In this tutorial, we covered how to develop and test an asynchronous API with FastAPI, Postgres, pytest, and Docker using Test-driven Development.

With Flask-like simplicity, Django-like batteries, and Go/Node-like performance, FastAPI is a powerful framework that makes it easy and fun to spin up RESTful APIs. Check your understanding by reviewing the objectives from the beginning of this tutorial and going through each of the challenges below.

Looking for some more challenges?

  1. Review the official tutorial. It's long but well worth a read.
  2. Implement async background tasks, database migrations, and auth.
  3. Abstract out the application configuration to a separate file.
  4. In a production environment, you'll probably want to stand up Gunicorn and let it manage Uvicorn. Review Running with Gunicorn and the Deployment guide for more info. Check out the official uvicorn-gunicorn-fastapi Docker image as well.
  5. Finally, check out the Test-Driven Development with FastAPI and Docker course as well as our other FastAPI courses for more!

You can find the source code in the fastapi-crud-async repo. Thanks for reading!

Featured Course

Test-Driven Development with FastAPI and Docker

In this course, you'll learn how to build, test, and deploy a text summarization service with Python, FastAPI, and Docker. The service itself will be exposed via a RESTful API and deployed to Heroku with Docker.

Buy Now $30 View Course
Featured Course

Test-Driven Development with FastAPI and Docker

In this course, you'll learn how to build, test, and deploy a text summarization service with Python, FastAPI, and Docker. The service itself will be exposed via a RESTful API and deployed to Heroku with Docker.

Buy Now $30 View Course