Database

Introduction

Any web application will most likely need to interact with a database. Expanse relies on SQLAlchemy for this purpose and make its integration seamless and intuitive.

Expanse supports natively the following databases:

• SQLite 3.36.0+
• MySQL 8.0+
• PostgreSQL 12.0+

However, if you need support for other databases supported by SQLAlchemy, you can easily extend the database manager.

Configuration

The configuration of your database is done through environment variables prefixed with DB_. Examples for the supported database are provided in the example .env.example file at the root of you application directory. You can edit and/or remove any value as you see fit.

The default database that will be used by Expanse is controlled via the DB_CONNECTION environment variable.

export DB_CONNECTION=sqlite

You can configure as many database connections as you want by following the following pattern:

export DB_CONNECTIONS__CONNECTION_NAME__DRIVER=postgresql

CONNECTION_NAME should be replaced by the actual name you want to give to your database connection.

Supported databases

SQLite

Since SQLite databases are single files on your filesystem, the only thing you need to configure is the path to your database via the DB_CONNECTIONS__SQLITE__DATABASE environment variable:

DB_CONNECTIONS__SQLITE__DRIVER=sqlite
DB_CONNECTIONS__SQLITE__DATABASE=/absolute/path/to/database.sqlite

To enable foreign key constraints for SQLite connections, you should set the DB_CONNECTIONS__SQLITE__FOREIGN_KEYS environment variable to true:

DB_CONNECTIONS__SQLITE__FOREIGN_KEYS=true

PostgreSQL

There are a few settings to configure to actually use a PostgreSQL database: these are the host, port, database, username and password.

They can be configured individually by using separate environment variables:

DB_CONNECTIONS__POSTGRESQL__DRIVER=postgresql
DB_CONNECTIONS__POSTGRESQL__HOST=127.0.0.1
DB_CONNECTIONS__POSTGRESQL__PORT=5432  # Optional since it's the default value
DB_CONNECTIONS__POSTGRESQL__DATABASE=expanse
DB_CONNECTIONS__POSTGRESQL__USERNAME=postgres
DB_CONNECTIONS__POSTGRESQL__PASSWORD=password

You can also combine them all by providing a complete URL instead:

DB_CONNECTIONS__POSTGRESQL__DRIVER=postgresql
DB_CONNECTIONS__POSTGRESQL__URL=postgresql://postgres:[email protected]:5432/expanse

Installing the necessary packages

SQLAlchemy needs a DBAPI package available to connect to the database. The one you need to install depends on how you configure the database, which DBAPI you want to use and the type of database connection you want to use (synchronous or asynchronous):

Synchronous

Asynchronous

The DBAPI can be specified through an environment variable or directly in the URL:

DB_CONNECTIONS__POSTGRESQL__DBAPI=psycopg
# Or
DB_CONNECTIONS__POSTGRESQL__URL=postgresql+psycopg://...

MySQL

There are a few settings to configure to actually use a MySQL database: these are the host, port, database, username, password and charset.

They can be configured individually by using separate environment variables:

DB_CONNECTIONS__MYSQL__DRIVER=mysql
DB_CONNECTIONS__MYSQL__HOST=127.0.0.1
DB_CONNECTIONS__MYSQL__PORT=3306  # Optional since it's the default value
DB_CONNECTIONS__MYSQL__DATABASE=expanse
DB_CONNECTIONS__MYSQL__USERNAME=root
DB_CONNECTIONS__MYSQL__PASSWORD=password
DB_CONNECTIONS__MYSQL__CHARSET=utf8mb4

You can also combine them all by providing a complete URL instead:

DB_CONNECTIONS__MYSQL__DRIVER=mysql
DB_CONNECTIONS__MYSQL__URL=mysql://root:[email protected]:3306/expanse?charset=utf8mb4

Installing the necessary packages

SQLAlchemy needs a DBAPI package available to connect to the database. The one you need to install depends on how you configure the database and which DBAPI you want to use and the type of database connection you want to use (synchronous or asynchronous):

Synchronous

Asynchronous

The DBAPI can be specified through an environment variable or directly in the URL:

DB_CONNECTIONS__MYSQL__DBAPI=pymysql
# Or
DB_CONNECTIONS__MYSQL__URL=mysql+pymysql://...

Executing SQL queries

Now that your database is configured, you can execute queries using a connection that you can retrieve from the database manager:

from expanse.database.database_manager import DatabaseManager
from expanse.view.view_factory import ViewFactory
from expanse.view.view import View


def list_users(db: DatabaseManager, view: ViewFactory) -> View:
    with db.connection() as connection:
        users = connection.execute(
            "select * from users where active = :active",
            {"active": True}
        )

        return view.make("users.index", {"users": users})

from expanse.database.database_manager import AsyncDatabaseManager
from expanse.view.view_factory import AsyncViewFactory
from expanse.view.view import View


async def list_users(db: AsyncDatabaseManager, view: AsyncViewFactory) -> View:
    async with db.connection() as connection:
        users = await connection.execute(
            "select * from users where active = :active",
            {"active": True}
        )

        return view.make("users.index", {"users": users})

Accessing the connection directly

Most of the time you will want to access the connection directly without having to go through the database manager, for that purpose you can type-hint a Connection (or AsyncConnection) that will automatically be injected:

from expanse.database.connection import Connection
from expanse.view.view import View
from expanse.view.view_factory import ViewFactory


async def list_users(
    connection: Connection,
    view: ViewFactory,
) -> View:
    users = connection.execute(
        "select * from users where active = :active",
        {"active": True}
    )

    return view.make("users.index", {"users": users})

from expanse.database.connection import AsyncConnection
from expanse.view.view import View
from expanse.view.view_factory import AsyncViewFactory


async def list_users(
    connection: AsyncConnection,
    view: AsyncViewFactory,
) -> View:
    users = await connection.execute(
        "select * from users where active = :active",
        {"active": True}
    )

    return view.make("users.index", {"users": users})

Accessing the connection this way makes sure that the connection will automatically be closed after the response is returned.

Using multiple database connections

If you have configured multiple database connections, you can specify which one you want when injecting it in you routes by annotating the Connection (or AsyncConnection) type. The name given must match one of the connections defined in your configuration.

from typing import Annotated
from expanse.database.connection import Connection
from expanse.view.view import View
from expanse.view.view_factory import ViewFactory


def list_users(
    connection: Annotated[Connection, "sqlite"],
    view: ViewFactory,
) -> View:
    users = connection.execute(
        "select * from users where active = :active",
        {"active": True}
    )

    return view.make("users.index", {"users": users})

from typing import Annotated
from expanse.database.connection import AsyncConnection
from expanse.view.view import View
from expanse.view.view_factory import AsyncViewFactory


def list_users(
    connection: Annotated[AsyncConnection, "sqlite"],
    view: AsyncViewFactory,
) -> View:
    users = connection.execute(
        "select * from users where active = :active",
        {"active": True}
    )

    return view.make("users.index", {"users": users})

Extending the database manager

If you need to use a database that is not natively supported by Expanse but is supported by SQLAlchemy, you can extend the database manager by adding a new database creator. A creator is a callable that must return a SQLAlchemy Engine instance.

Let's say you want to add support for MariaDB (note that it is technically not necessary to do so since a configured MySQL database can point to a MariaDB database and SQLAlchemy will accommodate the differences between the two). The first thing you want to do is to create a function that will take an instance of the database manager and a raw dictionary configuration, and return an engine:

from typing import Any
from sqlalchemy import Engine, URL, make_url
from expanse.database.database_manager import DatabaseManager


def create_mariadb_engine(
    db: DatabaseManager, raw_config: dict[str, Any]
) -> Engine:
    # Optionally validate the configuration (through a Pydantic model for instance)

    # Create a SQLAlchemy URL object
    if raw_url := raw_config.get("url"):
        url = make_url(raw_url)
    else:
        url = URL(...)

    engine = db.create_base_engine(url)

    # Configure the engine further if necessary

    return engine

from typing import Any
from sqlalchemy import Engine, URL, make_url
from expanse.database.database_manager import AsyncDatabaseManager


def create_mariadb_engine(
    db: AsyncDatabaseManager, raw_config: dict[str, Any]
) -> Engine:
    # Optionally validate the configuration (through a Pydantic model for instance)

    # Create a SQLAlchemy URL object
    if raw_url := raw_config.get("url"):
        url = make_url(raw_url)
    else:
        url = URL(...)

    engine = db.create_base_engine(url)

    # Configure the engine further if necessary

    return engine

Now you need to make the database manager aware of this creator and tell it for which database driver it should be used. This is typically done in a service provider and, for the sake of simplicity, this will be added to the default AppServiceProvider.

from expanse.database.database_manager import DatabaseManager
from expanse.support.service_provider import ServiceProvider


class AppServiceProvider(ServiceProvider):

    async def register(self) -> None:
        self._app.on_resolved(DatabaseManager, self.add_mariadb_creator)

    def add_mariadb_creator(self, db: DatabaseManager) -> None:
        db.extend("mariadb", create_mariadb_engine)

from expanse.database.database_manager import AsyncDatabaseManager
from expanse.support.service_provider import ServiceProvider


class AppServiceProvider(ServiceProvider):

    async def register(self) -> None:
        self._app.on_resolved(AsyncDatabaseManager, self.add_mariadb_creator)

    def add_mariadb_creator(self, db: AsyncDatabaseManager) -> None:
        db.extend("mariadb", create_mariadb_engine)

Here you can see that we are using the on_resolved() method of the application: the reason for that is because we do not need to extend the database manager if it's never resolved.