Postgres template in fixtures

conftest.py

import importlib
import os
from typing import Iterator

import alembic.command
import alembic.config
import psycopg
import pytest
import pytest_asyncio
from sqlalchemy import NullPool
from sqlalchemy.ext.asyncio import create_async_engine
from testcontainers.postgres import PostgresContainer

from myapp.server.database.sqlachemy import SqlachemyDbFactory


@pytest.fixture(scope="module")
def postgres() -> Iterator[PostgresContainer]:
    with PostgresContainer("postgres:16", driver="psycopg", dbname="template") as postgres:
        yield postgres


@pytest.fixture(scope="module")
def database(postgres: PostgresContainer) -> None:
    os.environ["MYAPP_DATABASE_ADDRESS"] = postgres.get_connection_url()
    cfg_path = os.path.join(importlib.resources.files("myapp.server.database"), "alembic.ini")
    config = alembic.config.Config(file_=cfg_path)
    alembic.command.upgrade(config, "head")
    yield
    alembic.command.downgrade(config, "base")
    del os.environ["MYAPP_DATABASE_ADDRESS"]


@pytest.fixture
def postgres_connection_url(postgres) -> Iterator[str]:
    with psycopg.connect(postgres.get_connection_url(driver=None), autocommit=True) as conn:
        cur = conn.cursor()
        cur.execute("CREATE DATABASE test ENCODING 'utf8' TEMPLATE template")
        postgres.dbname = "test"
        yield postgres.get_connection_url()
        cur.execute("DROP DATABASE test")
    postgres.dbname = "template"


@pytest_asyncio.fixture
async def conftest_db_factory(database, postgres_connection_url):
    engine = create_async_engine(postgres_connection_url, poolclass=NullPool)
    return SqlachemyDbFactory(engine)


@pytest_asyncio.fixture
async def conftest_db(conftest_db_factory):
    async with conftest_db_factory() as db:
        yield db