orionis.database.migrations.Migrator

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class Migrator(IMigrator): (source)

Constructor: Migrator(app, conn_manager)

View In Hierarchy

Runner applying and reverting the application migrations.

Migrations are discovered under database/migrations and applied in filename order, which is chronological because filenames carry a zero-padded sequential prefix. Each migration runs inside its own transaction together with its tracking record, so a failure never leaves the tracking table claiming a migration that did not fully apply.

Static Method	`__nextBatch`	Compute the batch number to assign to a new migration run.
Static Method	`__selectBatches`	Pick the recorded rows belonging to the batches being reverted.
Method	`__connection`	Resolve the connection migrations run against.
Async Method	`__deleteRecord`	Remove a migration's tracking record.
Method	`__discover`	Discover migration classes defined under the migrations directory.
Async Method	`__ensureMigrationsTable`	Create the migrations tracking table if it does not already exist.
Async Method	`__getRan`	Return every recorded migration, ordered by application order.
Method	`__init__`	Initialize the migrator.
Async Method	`__insertRecord`	Record a migration as applied.
Method	`__migrationsPath`	Return the directory holding migration files.
Async Method	`__revert`	Roll back the recorded migrations of the selected batches.
Async Method	`__runStep`	Run one migration and its tracking write atomically.
Async Method	`fresh`	Drop migration and seeder history, then apply every migration.
Async Method	`migrate`	Apply every migration that has not been run yet.
Async Method	`refresh`	Roll back migrations and immediately apply them again.
Async Method	`reset`	Revert every migration recorded on the connection.
Async Method	`rollback`	Revert the most recently applied migration batches.
Async Method	`status`	Report which migrations are applied and which are pending.
Class Variable	`__slots__`	Undocumented
Instance Variable	`__app`	Undocumented
Instance Variable	`__conn_manager`	Undocumented
Instance Variable	`__discovered_cache`	Undocumented

@staticmethod

def __nextBatch(ran: list[dict[str, Any]]) -> int: (source) ¶

Compute the batch number to assign to a new migration run.

Parameters
ran:`list of dict`	Previously recorded migrations.

Returns
`int`	`1` when nothing has run yet, otherwise the highest recorded batch plus one.

@staticmethod

def __selectBatches(ran: list[dict[str, Any]], steps: int | None) -> list[dict[str, Any]]: (source) ¶

Pick the recorded rows belonging to the batches being reverted.

Parameters
ran:`list of dict`	Every recorded migration in ascending `id` order.
steps:`int or None`	Number of most recent batches to select; `None` selects all of them.

Returns
`list of dict`	Selected rows, most recently applied first.

def __connection(self, name: str | None) -> IConnection: (source) ¶

Resolve the connection migrations run against.

Parameters
name:`str or None`	Named connection, or `None` for the default one.

Returns
`IConnection`	Connection bound to its configuration.

Raises
`ConnectionNotFoundException`	If the connection is not declared in the configuration.

async def __deleteRecord(self, connection: IConnection, name: str): (source) ¶

Remove a migration's tracking record.

Parameters
connection:`IConnection`	Connection used to run the statement.
name:`str`	Migration name to remove.

Returns
`None`	This method does not return a value.

def __discover(self) -> dict[str, type[Migration]]: (source) ¶

Discover migration classes defined under the migrations directory.

The result is cached on the instance after the first call, since the migrations directory does not change during a process lifetime and discovery involves filesystem traversal, module imports, and reflection. Only classes defined in a migration module are eligible; package initializers and reexports are ignored.

Returns
`dict of str to type`	Migration classes keyed by module file stem, ordered chronologically (filenames are zero-padded sequential ids).

Raises
`ValueError`	If multiple classes share a persisted migration name.

async def __ensureMigrationsTable(self, connection: IConnection): (source) ¶

Create the migrations tracking table if it does not already exist.

Parameters
connection:`IConnection`	Connection used to perform the schema operation.

Returns
`None`	This method does not return a value.

async def __getRan(self, connection: IConnection) -> list[dict[str, Any]]: (source) ¶

Return every recorded migration, ordered by application order.

Parameters
connection:`IConnection`	Connection used to run the query.

Returns
`list of dict`	Rows with `id`, `migration`, and `batch` keys.

def __init__(self, app: IApplication, conn_manager: IConnectionManager): (source) ¶

Initialize the migrator.

Parameters
app:`IApplication`	Application instance used to resolve the migrations directory.
conn_manager:`IConnectionManager`	Connection manager used to resolve the database connection.

Returns
`None`	This method does not return a value.

async def __insertRecord(self, connection: IConnection, name: str, batch: int): (source) ¶

Record a migration as applied.

Parameters
connection:`IConnection`	Connection used to run the statement.
name:`str`	Migration name to record.
batch:`int`	Batch number the migration belongs to.

Returns
`None`	This method does not return a value.

def __migrationsPath(self) -> Path: (source) ¶

Return the directory holding migration files.

Returns
`Path`	Absolute path to the `database_migrations` directory under the application base path.

async def __revert(self, steps: int | None, connection: str | None, events: MigrationEvents | None) -> list[str]: (source) ¶

Roll back the recorded migrations of the selected batches.

Parameters
steps:`int or None`	Number of most recent batches to revert; `None` reverts every recorded migration.
connection:`str or None`	Named connection to roll back, or `None` for the default.
events:`MigrationEvents or None`	Progress callbacks reported for each migration.

Returns
`list of str`	Names of the migrations reverted, most recent first.

Raises
`MigrationNotFoundException`	If a recorded migration has no matching migration file.

async def __runStep(self, connection: IConnection, name: str, migration_cls: type[Migration], batch: int | None, events: MigrationEvents, *, clear_seeders: bool = False): (source) ¶

Run one migration and its tracking write atomically.

Parameters
connection:`IConnection`	Connection the migration runs against.
name:`str`	Migration name.
migration_cls:`type of Migration`	Migration class to instantiate and run.
batch:`int or None`	Batch number to record when applying; `None` reverts the migration instead.
events:`MigrationEvents`	Progress callbacks reported for this migration.
clear_seeders:`bool, optional`	Drop seeder history when this is the last applied migration.

Returns
`None`	This method does not return a value.

Raises
`Exception`	Any exception raised by the migration propagates after the transaction is rolled back and the failure is reported.

async def fresh(self, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.fresh

Drop migration and seeder history, then apply every migration.

Unlike refresh, the tracking table itself is dropped, so the whole history is rebuilt as a single first batch. Seeder history is also removed when an older rollback left it behind.

Parameters
connection:`str or None, optional`	Named connection to rebuild, or `None` for the default.
events:`MigrationEvents or None, optional`	Progress callbacks reported for each migration.

Returns
`list of str`	Names of the migrations applied, in the order they ran.

async def migrate(self, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.migrate

Apply every migration that has not been run yet.

Parameters
connection:`str or None, optional`	Named connection to migrate, or `None` for the default one.
events:`MigrationEvents or None, optional`	Progress callbacks reported for each migration.

Returns
`list of str`	Names of the migrations applied, in the order they ran.

Raises
`Exception`	Any exception raised by a migration `up` method aborts the run and propagates to the caller.

async def refresh(self, steps: int | None = None, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.refresh

Roll back migrations and immediately apply them again.

Parameters
steps:`int or None, optional`	Number of batches to roll back first; `None` rolls back every recorded migration.
connection:`str or None, optional`	Named connection to refresh, or `None` for the default.
events:`MigrationEvents or None, optional`	Progress callbacks reported for each migration.

Returns
`list of str`	Names of the migrations re-applied, in the order they ran.

Raises
`ValueError`	If `steps` is not a positive integer.

async def reset(self, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.reset

Revert every migration recorded on the connection.

Parameters
connection:`str or None, optional`	Named connection to reset, or `None` for the default one.
events:`MigrationEvents or None, optional`	Progress callbacks reported for each migration.

Returns
`list of str`	Names of the migrations reverted, most recent first.

Raises
`MigrationNotFoundException`	If a recorded migration has no matching migration file.

async def rollback(self, steps: int = 1, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.rollback

Revert the most recently applied migration batches.

Parameters
steps:`int, optional`	Number of batches to roll back, starting from the most recent one. Defaults to `1`.
connection:`str or None, optional`	Named connection to roll back, or `None` for the default.
events:`MigrationEvents or None, optional`	Progress callbacks reported for each migration.

Returns
`list of str`	Names of the migrations reverted, most recent first.

Raises
`ValueError`	If `steps` is not a positive integer.
`MigrationNotFoundException`	If a recorded migration has no matching migration file.

async def status(self, *, connection: str | None = None) -> list[dict[str, Any]]: (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.status

Report which migrations are applied and which are pending.

Parameters
connection:`str or None, optional`	Named connection to inspect, or `None` for the default.

Returns
`list of dict`	One entry per discovered migration with `migration`, `ran` and `batch` keys, in chronological order.

__slots__: tuple[str, ...] = (source) ¶

overrides orionis.database.contracts.migrator.IMigrator.__slots__

Undocumented

__app = (source) ¶

Undocumented

__conn_manager = (source) ¶

Undocumented

__discovered_cache: dict[str, type[Migration]] | None = (source) ¶

Undocumented