Skip to content

Database setup

We have chosen to store the data the app needs in a postgres database, which we will need to connect to and manage the schema of via the app. To connect I like to use databases which is a great wrapper around fast lower level postgres drivers. It is installed via poetry,

Run this command in backend/

poetry add databases[postgresql]

As the app starts up it will need to connect to the database, ideally by setting up a connection pool. The connection pool ensures that each query can reuse a connection from the pool rather than having to create a new connection. This improves the user experience as connecting to the database is much slower than using an existing connection from the pool. The following code which should be added to backend/src/backend/lib/,

from databases import Database

async def create_database(url: str) -> Database:
    db = Database(url)
    await db.connect()
    return db

The following should then be added to backend/src/backend/ to make a db app attribute available,

from backend.lib.database import create_database

def create_app() -> Quart:

    async def startup() -> None:
        app.db = await create_database(os.environ["DATABASE_URL"])

    async def shutdown() -> None:
        await app.db.disconnect()

As the database changes depending on the environment we'll need to make the following changes to the environment configuration files,


Schema management

The schema (structure of the database) needs to be set and then changed over time. Each change to the schema, and any accompanying data change, is termed a migration.

I find migrations work best if they are considered forward only, i.e. changes are sequential and take the database from a state into a new state. For this reason I manage migrations as numbered scripts, whereby migration 0 is the initial setup and 1 then next etc...

To manage the migrations we need to keep track of which migrations have been applied to the database, we can do this by creating a single rowed table to store the last migration by adding the following to backend/src/backend/,

async def create_database(url: str) -> Database:
    await _create_migration_table(db)
    return db

async def _create_migration_table(db: Database) -> None:
    await db.execute(
        """CREATE TABLE IF NOT EXISTS schema_migration (
               onerow_id BOOL PRIMARY KEY DEFAULT TRUE,
               version INTEGER NOT NULL,

               CONSTRAINT onerow_uni CHECK (onerow_id)
    await db.execute(
        """INSERT INTO schema_migration (version)

The migrations themselves the migrations must be run, but only one once (to prevent data corruption). To do this the following code should be placed in backend/src/backend/lib/,

import importlib.util

from databases import Database

async def create_database(url: str) -> Database:
    await _setup_schema(db)
    return db

async def _setup_schema(db: Database) -> int:
    migration = await db.fetch_val("SELECT version FROM schema_migration")
    initial_migration = migration
    while True:
        migration += 1
            await _run_migration(db, migration)
        except FileNotFoundError:
    return initial_migration

async def _run_migration(db: Database, migration: int) -> None:
    spec = importlib.util.spec_from_file_location(
        f"db_{migration}", f"db/{migration}.py"
    module = importlib.util.module_from_spec(spec)
    await module.migrate(db)
    await db.execute(
        "UPDATE schema_migration SET version = :version",
        values={"version": migration},

which will run migrations in sequential order as saved in python files in the backend/db folder, for example the first migration must be backend/db/0.sql.