Build a FastAPI Server

In this tutorial, we’ll build a production-ready FastAPI server together. The full functional example is available here.

Our application stack will look like this:

Start a New Project

Instead of pip, let’s use the shiny Poetry to manage our project. Follow the link to install Poetry, and create our new project in an empty directory:

$ mkdir gino-fastapi-demo
$ cd gino-fastapi-demo
$ git init
$ poetry init

Then follow the Poetry guide to finish the initialization - you may say “no” to the interactive dependency creation, because we will add them manually. It is okay to use the default values in the other steps of the guide, and make sure the package name remains gino-fastapi-demo.

Add Dependencies

FastAPI is built on top of the Starlette framework, so we shall use the GINO extension for Starlette. Simply run:

$ poetry add gino[starlette]

Then let’s add FastAPI, together with the lightning-fast ASGI server Uvicorn, and Gunicorn as a production application server:

$ poetry add fastapi uvicorn gunicorn

For database migration, we’ll use Alembic. Because it uses normal DB-API, we need psycopg here too:

$ poetry add alembic psycopg2

At last, let’s add pytest in the development environment for testing. We also want to add the requests library to use the Starlette TestClient:

$ poetry add -D pytest requests

Hint

With the steps above, Poetry will automatically create a virtualenv for you behind the scene, and all the dependencies are installed there. We will assume using this for the rest of the tutorial. But you’re free to create your own virtualenv, and Poetry will honor it when it’s activated.

That’s all, this is my pyproject.toml created by Poetry, yours should look similar:

[tool.poetry]
name = "gino-fastapi-demo"
version = "0.1.0"
description = ""
authors = ["Fantix King <fantix.king@gmail.com>"]

[tool.poetry.dependencies]
python = "^3.8"
gino = {version = "^1.0", extras = ["starlette"]}
fastapi = "^0.54.1"
uvicorn = "^0.11.3"
gunicorn = "^20.0.4"
alembic = "^1.4.2"
psycopg2 = "^2.8.5"

[tool.poetry.dev-dependencies]
pytest = "^5.4.1"
requests = "^2.23.0"

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

And there’s also an auto-generated poetry.lock file with the frozen versions. The directory layout should look like the diagram on the right. Now let’s add the two files to the Git repository (we will skip showing these git operations in future steps):

$ git add pyproject.toml poetry.lock
$ git commit -m 'add project dependencies'

Write a Simple Server

Now let’s write some Python code.

We’ll create an extra src directory to include all the Python files, as demonstrated in the diagram below. This is known as the “src layout” providing a cleaner hierarchy.

The root Python package of our project is named as gino_fastapi_demo, under which we will create two Python modules:

• asgi as the ASGI entry point - we’ll feed it to the ASGI server

• main to initialize our server

Here’s main.py:

from fastapi import FastAPI

def get_app():
    app = FastAPI(title="GINO FastAPI Demo")
    return app

And we’ll simply instantiate our application in asgi.py:

from .main import get_app

app = get_app()

Then run poetry install to link our Python package into the PYTHONPATH in development mode. We’ll be able to start a Uvicorn development server after that:

$ poetry install
Installing dependencies from lock file

No dependencies to install or update

  - Installing gino-fastapi-demo (0.1.0)

$ poetry run uvicorn gino_fastapi_demo.asgi:app --reload
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [53010]
INFO:     Started server process [53015]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

The --reload option enables Uvicorn to automatically reload the server for us when the Python source code is updated. Now access http://127.0.0.1:8000/docs to see the Swagger UI of our new FastAPI server.

Hint

As mentioned previously, if you’re in your own virtualenv, the command poetry run uvicorn can be simplified as just uvicorn.

poetry run is a convenient shortcut to run the following command in the virtualenv managed by Poetry.

Add GINO Extension

Now let’s add GINO to our server.

First of all, we need a way to configure the database. In this tutorial, we’ll use the configuration system from Starlette. Add src/gino_fastapi_demo/config.py as follows:

from sqlalchemy.engine.url import URL, make_url
from starlette.config import Config
from starlette.datastructures import Secret

config = Config(".env")

DB_DRIVER = config("DB_DRIVER", default="postgresql")
DB_HOST = config("DB_HOST", default=None)
DB_PORT = config("DB_PORT", cast=int, default=None)
DB_USER = config("DB_USER", default=None)
DB_PASSWORD = config("DB_PASSWORD", cast=Secret, default=None)
DB_DATABASE = config("DB_DATABASE", default=None)
DB_DSN = config(
    "DB_DSN",
    cast=make_url,
    default=URL(
        drivername=DB_DRIVER,
        username=DB_USER,
        password=DB_PASSWORD,
        host=DB_HOST,
        port=DB_PORT,
        database=DB_DATABASE,
    ),
)
DB_POOL_MIN_SIZE = config("DB_POOL_MIN_SIZE", cast=int, default=1)
DB_POOL_MAX_SIZE = config("DB_POOL_MAX_SIZE", cast=int, default=16)
DB_ECHO = config("DB_ECHO", cast=bool, default=False)
DB_SSL = config("DB_SSL", default=None)
DB_USE_CONNECTION_FOR_REQUEST = config(
    "DB_USE_CONNECTION_FOR_REQUEST", cast=bool, default=True
)
DB_RETRY_LIMIT = config("DB_RETRY_LIMIT", cast=int, default=1)
DB_RETRY_INTERVAL = config("DB_RETRY_INTERVAL", cast=int, default=1)

This config file will load from environment variable first, if not found then from a file named .env from current path (usually the project root directory), and at last use the default value defined above. For example, you can either overwrite in CLI directly like this:

$ DB_HOST=localhost DB_USER=postgres poetry run uvicorn gino_fastapi_demo.asgi:app --reload

Or set them in the file .env (this file must not be committed into Git, remember to add it to .gitignore):

DB_HOST=localhost
DB_USER=postgres

Now it’s time to create a PostgreSQL database and set the connection variables correctly here. This is usually something like createdb yourdbname, but it may vary across different platforms, so we won’t cover this part in this tutorial.

Tip

Alternatively, you could also set DB_DSN to for example postgresql://user:password@localhost:5432/dbname to override the other individual config values like DB_HOST defined before DB_DSN.

If defined, DB_DSN always have the higher priority over the individual ones, regardless of where they are defined - even if DB_HOST is defined in environment variable and DB_DSN is defined in .env file, DB_HOST is still ignored. Default value doesn’t count.

Then, create a new Python sub-package gino_fastapi_demo.models to encapsulate database-related code, and add the code below to src/gino_fastapi_demo/models/__init__.py:

from gino.ext.starlette import Gino

from .. import config

db = Gino(
    dsn=config.DB_DSN,
    pool_min_size=config.DB_POOL_MIN_SIZE,
    pool_max_size=config.DB_POOL_MAX_SIZE,
    echo=config.DB_ECHO,
    ssl=config.DB_SSL,
    use_connection_for_request=config.DB_USE_CONNECTION_FOR_REQUEST,
    retry_limit=config.DB_RETRY_LIMIT,
    retry_interval=config.DB_RETRY_INTERVAL,
)

At last, modify src/gino_fastapi_demo/main.py to install the GINO extension:

 from fastapi import FastAPI
+
+from .models import db

 def get_app():
     app = FastAPI(title="GINO FastAPI Demo")
+    db.init_app(app)
     return app

Saving the file, you should see the Uvicorn server reloads our changes and connects to the database:

WARNING:  Detected file change in 'src/gino_fastapi_demo/main.py'. Reloading...
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [63562]
INFO:     Started server process [63563]
INFO:     Waiting for application startup.
INFO:     Connecting to the database: postgresql://fantix:***@localhost
INFO:     Database connection pool created: <asyncpg.pool.Pool max=16 min=1 cur=1 use=0>
INFO:     Application startup complete.

Create Models and API

It’s time to implement the API now. Let’s say we are building a user management service, through which we could add users, list users and delete users.

First of all, we need a database table users to store the data, mapped to a GINO model named User. We shall add the model in gino_fastapi_demo.models.users:

from . import db

class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.BigInteger(), primary_key=True)
    nickname = db.Column(db.Unicode(), default="unnamed")

The model definition is simple enough to explain itself.

Then we only have to use it properly in the API implementation, for which we’ll create a new Python sub-package gino_fastapi_demo.views, and a new module gino_fastapi_demo.views.users as follows:

from fastapi import APIRouter
from pydantic import BaseModel

from ..models.users import User

router = APIRouter()

@router.get("/users/{uid}")
async def get_user(uid: int):
    user = await User.get_or_404(uid)
    return user.to_dict()

class UserModel(BaseModel):
    name: str

@router.post("/users")
async def add_user(user: UserModel):
    rv = await User.create(nickname=user.name)
    return rv.to_dict()

@router.delete("/users/{uid}")
async def delete_user(uid: int):
    user = await User.get_or_404(uid)
    await user.delete()
    return dict(id=uid)

def init_app(app):
    app.include_router(router)

The APIRouter holds our new APIs locally, and init_app is used to integrate it into our FastAPI application. Here we want some inversion of control: let’s make the APIs plugable, so that we don’t have to import all possible future views manually. We shall use the Entry Points feature to load the dependencies. Add this code below to gino_fastapi_demo.main:

import logging
from importlib.metadata import entry_points

logger = logging.getLogger(__name__)

def load_modules(app=None):
    for ep in entry_points()["gino_fastapi_demo.modules"]:
        logger.info("Loading module: %s", ep.name)
        mod = ep.load()
        if app:
            init_app = getattr(mod, "init_app", None)
            if init_app:
                init_app(app)

Hint

If you’re running Python < 3.8, you’ll need this importlib-metadata backport.

And call it in our application factory:

 def get_app():
     app = FastAPI(title="GINO FastAPI Demo")
     db.init_app(app)
+    load_modules(app)
     return app

Finally, define the entry points in pyproject.toml following the Poetry document for plugins:

[tool.poetry.plugins."gino_fastapi_demo.modules"]
"users" = "gino_fastapi_demo.views.users"

Run poetry install again to activate the entry points - you may need to restart the Uvicorn development server manually, as the reloader cannot capture the changes we made to pyproject.toml.

Now you should be able to see the 3 new APIs on the Swagger UI. But none of them works, because we still haven’t created the database tables.

Integrate with Alembic

To get started with Alembic, run this command in the project root directory:

$ poetry run alembic init migrations

This will generate a new directory migrations where Alembic will store database migration revisions. At the same time, an alembic.ini file is created in the project root directory. Let’s simply add all of them to Git control.

For Alembic to use our data models defined with GINO (and of course the database config), we need to modify migrations/env.py to connect with the GINO instance:

 # add your model's MetaData object here
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
-target_metadata = None
+from gino_fastapi_demo.config import DB_DSN
+from gino_fastapi_demo.main import db, load_modules
+
+load_modules()
+config.set_main_option("sqlalchemy.url", str(DB_DSN))
+target_metadata = db

Then create our first migration revision with:

$ poetry run alembic revision --autogenerate -m 'add users table'
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'users'
  Generating migrations/versions/32c0feba61ea_add_users_table.py ...  done

The generated revision file should roughly look like this:

def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.BigInteger(), nullable=False),
        sa.Column("nickname", sa.Unicode(), nullable=True),
        sa.PrimaryKeyConstraint("id"),
    )

def downgrade():
    op.drop_table("users")

Hint

Whenever there is a change to the database schema in the future, just modify the GINO models and run alembic revision --autogenerate again to generate new revisions to track the change. Remember to review the revision file - you may want to adjust it.

Eventually, let’s apply this migration, by upgrading to the latest revision:

$ poetry run alembic upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 32c0feba61ea, add users table

Now all the APIs should be fully operational, try with the Swagger UI.

Write the Tests

In order not to break our development database with running tests, let’s create a separate database to run tests. Apply this change to gino_fastapi_demo.config:

 config = Config(".env")

+TESTING = config("TESTING", cast=bool, default=False)

 DB_DRIVER = config("DB_DRIVER", default="postgresql")
 DB_HOST = config("DB_HOST", default=None)
 DB_PORT = config("DB_PORT", cast=int, default=None)
 DB_USER = config("DB_USER", default=None)
 DB_PASSWORD = config("DB_PASSWORD", cast=Secret, default=None)
 DB_DATABASE = config("DB_DATABASE", default=None)
+if TESTING:
+    if DB_DATABASE:
+        DB_DATABASE += "_test"
+    else:
+        DB_DATABASE = "gino_fastapi_demo_test"
 DB_DSN = config(

Hint

You need to run createdb to actually create the database. If you have set DB_DATABASE in .env - e.g. DB_DATABASE=mydb, the name of the testing database should be mydb_test. Or else, gino_fastapi_demo_test.

Then, let’s create our pytest fixture in tests/conftest.py:

import pytest
from alembic.config import main
from starlette.config import environ
from starlette.testclient import TestClient

environ["TESTING"] = "TRUE"

@pytest.fixture
def client():
    from gino_fastapi_demo.main import db, get_app

    main(["--raiseerr", "upgrade", "head"])

    with TestClient(get_app()) as client:
        yield client

    main(["--raiseerr", "downgrade", "base"])

This fixture creates all the database tables before running the test, yield a Starlette TestClient, and drop all the tables with all the data after the test to maintain a clean environment for the next test.

Here’s a sample test in tests/test_users.py:

import uuid

def test_crud(client):
    # create
    nickname = str(uuid.uuid4())
    r = client.post("/users", json=dict(name=nickname))
    r.raise_for_status()

    # retrieve
    url = f"/users/{r.json()['id']}"
    assert client.get(url).json()["nickname"] == nickname

    # delete
    client.delete(url).raise_for_status()
    assert client.get(url).status_code == 404

Then run the test:

$ poetry run pytest
=========================== test session starts ===========================
platform darwin -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1
rootdir: gino-fastapi-demo
collected 1 item

tests/test_users.py .                                               [100%]

============================ 1 passed in 1.21s ============================

Notes for Production

Given the popularity of Docker/Kubernetes, we’ll build a Dockerfile for our demo:

FROM python:3.8-alpine as base

FROM base as builder
RUN apk add --no-cache gcc musl-dev libffi-dev openssl-dev make postgresql-dev
RUN pip install poetry
COPY . /src/
WORKDIR /src
RUN python -m venv /env && . /env/bin/activate && poetry install

FROM base
RUN apk add --no-cache postgresql-libs
COPY --from=builder /env /env
COPY --from=builder /src /src
WORKDIR /src
CMD ["/env/bin/gunicorn", "gino_fastapi_demo.asgi:app", "-b", "0.0.0.0:80", "-k", "uvicorn.workers.UvicornWorker"]

In this Dockerfile, we used 2 phases to separate the building from the production image to reduce target artifact size. Also, we are using Gunicorn with UvicornWorker from Uvicorn as the worker class for best production reliability.

Let’s review what we have in the project.

This is the end of the tutorial to build a demo. Below is an incomplete checklist to go live:

• Set DB_RETRY_LIMIT to a larger number to allow staring the application server before the database is fully ready.

• Implement the same retry logic in migrations/env.py so that Alembic gets the same functionality.

• Enable DB_SSL if needed.

• Write a docker-compose.yml for other developers to get a quick taste or even use it for development.

• Enable CI, install pytest-cov and use --cov-fail-under to guarantee coverage.

• Integrate static code analysis tools and security/CVE checking tools.

• Automate Alembic upgrade properly - e.g. after new version is deployed.

• Be aware of the common security attacks like CSRF, XSS, etc.

• Write load tests.

Again, the source code of the demo is available here, and the source of this tutorial is here. Please feel free to submit PRs to fix issues or add your thoughts. Happy hacking!