ORIONIS API REFERENCE

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.

class documentation

Contract for the database migration runner.

Implementations discover migration files under the application's database/migrations directory, track which ones have already been applied, and apply or revert them in chronological order against any configured connection.

Async Method fresh Drop the tracking table and apply every migration from scratch.
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
async def fresh(self, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source)

Drop the tracking table and apply every migration from scratch.

Parameters
connection:str or None, optionalNamed connection to rebuild, or None for the default.
events:MigrationEvents or None, optionalProgress callbacks reported for each migration.
Returns
list of strNames of the migrations applied, in the order they ran.
async def migrate(self, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source)

Apply every migration that has not been run yet.

Parameters
connection:str or None, optionalNamed connection to migrate, or None for the default one.
events:MigrationEvents or None, optionalProgress callbacks reported for each migration.
Returns
list of strNames of the migrations applied, in the order they ran. Empty when there is nothing pending.
Raises
ExceptionAny exception raised by a migration's 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)

Roll back migrations and immediately apply them again.

Parameters
steps:int or None, optionalNumber of batches to roll back first; None rolls back every recorded migration.
connection:str or None, optionalNamed connection to refresh, or None for the default.
events:MigrationEvents or None, optionalProgress callbacks reported for each migration.
Returns
list of strNames of the migrations re-applied, in the order they ran.
Raises
ValueErrorIf steps is not a positive integer.
async def reset(self, *, connection: str | None = None, events: MigrationEvents | None = None) -> list[str]: (source)

Revert every migration recorded on the connection.

Parameters
connection:str or None, optionalNamed connection to reset, or None for the default one.
events:MigrationEvents or None, optionalProgress callbacks reported for each migration.
Returns
list of strNames of the migrations reverted, most recent first.
Raises
MigrationNotFoundExceptionIf 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)

Revert the most recently applied migration batches.

Parameters
steps:int, optionalNumber of batches to roll back, starting from the most recent one. Defaults to 1.
connection:str or None, optionalNamed connection to roll back, or None for the default.
events:MigrationEvents or None, optionalProgress callbacks reported for each migration.
Returns
list of strNames of the migrations reverted, most recent first. Empty when there is nothing to revert.
Raises
ValueErrorIf steps is not a positive integer.
MigrationNotFoundExceptionIf a recorded migration has no matching migration file.
async def status(self, *, connection: str | None = None) -> list[dict[str, Any]]: (source)

Report which migrations are applied and which are pending.

Parameters
connection:str or None, optionalNamed connection to inspect, or None for the default.
Returns
list of dictOne entry per discovered migration with migration, ran and batch keys, in chronological order.
__slots__: tuple = (source)

Undocumented