orionis.database.seeders.SeederRunner

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 SeederRunner: (source)

Constructor: SeederRunner(app, conn_manager)

View In Hierarchy

Apply new seeders exactly once on the selected database connection.

Files under database/seeders are discovered in lexical filename order. The application container builds each seeder so constructors can receive dependencies. Each seeder and its tracking row commit in the same transaction. The unique tracking key acts as a database-level claim when runs overlap.

Method	`__discover`	Find local Seeder subclasses in deterministic filename order.
Async Method	`__ensureTrackingTable`	Create the tracking table, tolerating a concurrent creator.
Async Method	`__getRan`	Read committed tracking rows in insertion order.
Method	`__init__`	Initialize the runner with the application and connection manager.
Async Method	`__isRecorded`	Check whether another transaction committed this seeder.
Async Method	`__runStep`	Claim, run and commit one seeder and its tracking row.
Method	`__seedersPath`	Return the application's configured seeder directory.
Async Method	`seed`	Run every discovered seeder that has no committed tracking row.
Class Variable	`__slots__`	Undocumented
Instance Variable	`__app`	Undocumented
Instance Variable	`__conn_manager`	Undocumented
Instance Variable	`__discovered_cache`	Undocumented

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

Find local Seeder subclasses in deterministic filename order.

Returns
`dict of str to type of Seeder`	Seeder classes keyed by persisted filename stem.

Raises
`ValueError`	If more than one class resolves to the same persisted name.

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

Create the tracking table, tolerating a concurrent creator.

Parameters
connection:`IConnection`	Selected database connection.

Returns
`None`	The table is available for tracking operations.

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

Read committed tracking rows in insertion order.

Parameters
connection:`IConnection`	Connection holding the tracking table.

Returns
`list of dict`	Tracking rows ordered by primary key.

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

Initialize the runner with the application and connection manager.

Parameters
app:`IApplication`	Application providing the seeder directory.
conn_manager:`IConnectionManager`	Manager resolving configured database connections.

Returns
`None`	Complete the documented operation without returning a value.

async def __isRecorded(self, connection: IConnection, name: str) -> bool: (source) ¶

Check whether another transaction committed this seeder.

Parameters
connection:`IConnection`	Connection holding the tracking table.
name:`str`	Seeder filename stem.

Returns
`bool`	Whether the tracking row is committed.

async def __runStep(self, connection: IConnection, name: str, seeder_cls: type[Seeder], batch: int, events: SeederEvents) -> bool: (source) ¶

Claim, run and commit one seeder and its tracking row.

Parameters
connection:`IConnection`	Connection shared by the seeder and tracking transaction.
name:`str`	Persisted seeder filename stem.
seeder_cls:`type of Seeder`	Class to instantiate and execute.
batch:`int`	Batch assigned to this invocation.
events:`SeederEvents`	Progress callbacks.

Returns
`bool`	Whether this call executed the seeder.

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

Return the application's configured seeder directory.

Returns
`Path`	Absolute path under `database/seeders`.

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

Run every discovered seeder that has no committed tracking row.

Parameters
connection:`str or None, optional`	Named database connection, or the configured default.
events:`SeederEvents or None, optional`	Progress callbacks for executed seeders.

Returns
`list of str`	Seeder names completed by this call, in execution order.

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

Undocumented

__app = (source) ¶

Undocumented

__conn_manager = (source) ¶

Undocumented

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

Undocumented