This content originally appeared on HackerNoon and was authored by Artem 8x
Following my previous post about setting a function-level database setup, which is a junior-level solution, we’ll be looking at a session-level database migration setup.
When initiating a DB-coupled application, one of the initial goals is to set a DB connection function or class for spawning a reliable asynchronous connection with our DB.
Let us cover the most important parts of our setup.
The DB session manager class:
from sqlalchemy.ext.asyncio import (
async_sessionmaker,
create_async_engine,
AsyncEngine,
AsyncSession,
)
class DBSessionManager:
def __init__(self, postgres_dsn: str):
self._engine: AsyncEngine = create_async_engine(url=postgres_dsn)
self._async_sesionmaker = async_sessionmaker(
bind=self._engine, expire_on_commit=False
)
@property
def asessionmaker(self) -> async_sessionmaker[AsyncSession]:
return self._async_sesionmaker
async def close(self):
await self._engine.dispose()
\ With the same set of models:
class Product(Base):
__tablename__ = 'product'
id: Mapped[UUID] = mapped_column(
type_=types.UUID,
primary_key=True,
server_default=text('gen_random_uuid()'),
)
name: Mapped[str] = mapped_column(
type_=types.VARCHAR(100), server_default=text("''")
)
created_at: Mapped[timestamp] = mapped_column(
type_=types.TIMESTAMP,
server_default=text('NOW()'),
)
class Review(Base):
__tablename__ = 'review'
id: Mapped[UUID] = mapped_column(
type_=types.UUID,
primary_key=True,
server_default=text('gen_random_uuid()'),
)
content: Mapped[str] = mapped_column(
type_=types.VARCHAR(1000), server_default=text("''")
)
rating: Mapped[int] = mapped_column(type_=types.DECIMAL(2, 1))
created_at: Mapped[timestamp] = mapped_column(
type_=types.TIMESTAMP,
server_default=text('NOW()'),
)
\
:::info Note: the test setup file is still the bottleneck of our test environment setup.
:::
\
The Essence of the Fixture Setup
The key fixtures to implement in a Python application with a database connection include:
- Database creation and teardown
- Schema version upgrades and downgrades
- Connection setup and database content reset
\
@pytest_asyncio.fixture(scope='session')
async def create_test_db(os_environ_patch):
test_db_name = 'example_db_test'
engine = create_async_engine(
os.environ['POSTGRES_DSN_ORIGINAL'],
isolation_level='AUTOCOMMIT',
)
create_db_op = text(f'CREATE DATABASE {test_db_name}')
drop_db_op = text(f'DROP DATABASE IF EXISTS {test_db_name} WITH (FORCE)')
async with engine.begin() as conn:
await conn.execute(create_db_op)
yield
async with engine.connect() as conn:
await conn.execute(drop_db_op)
@pytest.fixture(scope='session')
def migrate_db(create_test_db):
config = Config('alembic.ini')
test_db_url = os.environ['POSTGRES_DSN']
config.set_main_option('sqlalchemy.url', test_db_url)
command.upgrade(config, 'head')
yield
command.downgrade(config, 'base')
@pytest_asyncio.fixture
async def db(migrate_db) -> AsyncGenerator[DBSessionManager, None]:
postgres_dsn = os.environ['POSTGRES_DSN']
db_manager = DBSessionManager(postgres_dsn)
yield db_manager
target_metadata = Base.metadata
tables = target_metadata.tables.keys()
all_tables_str = ', '.join(f'"{t}"' for t in tables)
async with db_manager.asessionmaker() as s:
await s.execute(text(f'TRUNCATE TABLE {all_tables_str} CASCADE'))
await s.commit()
await db_manager.close()
\ Now, let’s zoom in on the most important parts.
Migrations
@pytest.fixture(scope='session')
def migrate_db(create_test_db):
The above lets us run through the migration step only once per session.
Tables truncation
Here, the DB fixture is relying on the session manager to execute custom SQL transactions.
\
target_metadata = Base.metadata
tables = target_metadata.tables.keys() # dict_keys(['product', 'review'])
all_tables_str = ', '.join(f'"{t}"' for t in tables) # '"product", "review"'
The code above extracts the registered tables to the comma-separated and quotation marks-wrapped representation.
After that, TRUNCATE TABLE {all_tables_str} CASCADE will delete all the records in the tables using cascade mode by deleting records in the constraints-dependent tables.
The final step is to dispose of the DB manager instance
await db_manager.close()
This way, we are ensured the migration process is set up correctly within our Python application.
\
This content originally appeared on HackerNoon and was authored by Artem 8x
Artem 8x | Sciencx (2025-11-10T14:16:27+00:00) How to Set Up Session-Level Database Migrations in Python. Retrieved from https://www.scien.cx/2025/11/10/how-to-set-up-session-level-database-migrations-in-python/
Please log in to upload a file.
There are no updates yet.
Click the Upload button above to add an update.