orionis.database.schema.schema.Schema

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

Constructor: Schema(conn_manager)

View In Hierarchy

Undocumented

Method	`__addTimestampColumns`	Expand a `Timestamps` marker into its two nullable columns.
Method	`__buildTable`	Build the table definition from a name and its definitions.
Method	`__classifyDefinition`	Route a single schema definition into the shared bucket.
Method	`__collectDefinitions`	Group heterogeneous schema definitions into constructor kwargs.
Method	`__init__`	Initialize the Schema instance.
Method	`__parseTableName`	Parse the table name to extract schema and table components.
Method	`__resolveConnection`	Resolve the connection bound to the configured connection name.
Method	`__setPrimaryKey`	Assign the primary key columns, rejecting duplicate definitions.
Async Method	`_createTable`	Compile and execute the `CREATE TABLE` for `name`.
Method	`connection`	Set the connection name for schema operations.
Method	`create`	Start a fluent table declaration block.
Async Method	`createFromDefinition`	Create a table from its reusable, versioned definition.
Async Method	`createFromModel`	Create the current schema declared by a concrete model.
Async Method	`drop`	Drop an existing table.
Class Variable	`__slots__`	Undocumented
Instance Variable	`__conn_manager`	Undocumented
Instance Variable	`__connection_name`	Undocumented
Instance Variable	`__connection_selected`	Undocumented

def __addTimestampColumns(self, definition: Timestamps, bucket: DefinitionBucket): (source) ¶

Expand a Timestamps marker into its two nullable columns.

Parameters
definition:`Timestamps`	The timestamps marker carrying the `timezone` flag.
bucket:`DefinitionBucket`	Accumulator updated in place with the new columns.

Returns
`None`	The `bucket` accumulator is mutated in place.

def __buildTable(self, name: str, definitions: tuple[SchemaDefinition, ...]) -> TableDefinition: (source) ¶

Build the table definition from a name and its definitions.

Parameters
name:`str`	The name of the table to build, possibly `schema.table`.
definitions:`tuple`	Schema definitions collected for the table.

Returns
`TableDefinition`	The complete table definition ready for compilation.

Route a single schema definition into the shared bucket.

Parameters
definition:`ColumnDefinition \| Comment \| ForeignKey \| Index \|`	PrimaryKey \| Unique The schema definition to classify.
bucket:`DefinitionBucket`	Accumulator updated in place with the classified data.

Returns
`None`	The `bucket` accumulator is mutated in place.

def __collectDefinitions(self, definitions: tuple[SchemaDefinition, ...]) -> dict[str, object]: (source) ¶

Group heterogeneous schema definitions into constructor kwargs.

Parameters
definitions:`tuple`	Column, constraint, index, and comment definitions to sort.

Returns
`dict[str, object]`	Keyword arguments accepted by `TableDefinition`.

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

Initialize the Schema instance.

Parameters
conn_manager:`IConnectionManager`	The connection manager for database operations.

Returns
`None`	Undocumented

def __parseTableName(self, name: str) -> tuple[str | None, str]: (source) ¶

Parse the table name to extract schema and table components.

Parameters
name:`str`	The full table name, which may include a schema prefix.

Returns
`tuple[str \| None, str]`	A tuple containing the schema name (or None if not specified) and the table name.

def __resolveConnection(self) -> IConnection: (source) ¶

Resolve the connection bound to the configured connection name.

Returns
`IConnection`	The connection to use for schema operations.

def __setPrimaryKey(self, column_names: list[str], kwargs: dict[str, object]): (source) ¶

Assign the primary key columns, rejecting duplicate definitions.

Parameters
column_names:`list[str]`	Names of the columns composing the primary key.
kwargs:`dict[str, object]`	Accumulated `TableDefinition` keyword arguments, updated in place.

Returns
`None`	The `kwargs` mapping is mutated in place.

Raises
`ValueError`	If a primary key was already defined.

async def _createTable(self, name: str, definitions: tuple[SchemaDefinition, ...]) -> bool: (source) ¶

Compile and execute the CREATE TABLE for name.

Parameters
name:`str`	The name of the table to create.
definitions:`tuple`	Schema definitions collected for the table.

Returns
`bool`	`True` when the table is created without errors.

def connection(self, name: str | None = None) -> Self: (source) ¶

overrides orionis.database.contracts.schema.ISchema.connection

Set the connection name for schema operations.

Parameters
name:`str \| None, optional`	The connection name to use, by default None.

Returns
`Self`	The current Schema instance for method chaining.

Raises
`ValueError`	If connection name has already been defined.

def create(self, name: str) -> TableCreation: (source) ¶

overrides orionis.database.contracts.schema.ISchema.create

Start a fluent table declaration block.

async with schema.create(name) as table: yields a ~orionis.database.schema.blueprint.Blueprint so columns can be declared fluently (table.string("username"), table.timestamps(), ...); the table is created once the block exits without raising.

Parameters
name:`str`	The name of the table to create. If the table belongs to a non-default schema, use the `schema.table` format.

Returns
`TableCreation`	Async context manager that collects definitions and creates the table after a successful block.

async def createFromDefinition(self, definition: TableDefinition) -> bool: (source) ¶

overrides orionis.database.contracts.schema.ISchema.createFromDefinition

Create a table from its reusable, versioned definition.

Parameters
definition:`TableDefinition`	Schema shared with a model through `table_definition`. Historical migrations should import a versioned definition that remains unchanged when the model adopts a newer version.

Returns
`bool`	Whether table creation completed successfully.

async def createFromModel(self, model: type[Model]) -> bool: (source) ¶

overrides orionis.database.contracts.schema.ISchema.createFromModel

Create the current schema declared by a concrete model.

Use this operation for bootstrap and temporary databases. A historical migration should call createFromDefinition with a versioned table definition so later model changes do not alter migration replay.

Parameters
model:`type of Model`	Concrete model providing table and connection metadata. An explicit `connection` selection overrides the model's connection.

Returns
`bool`	Whether table creation completed successfully.

Raises
`TypeError`	If the argument is not a concrete model class.

async def drop(self, name: str) -> bool: (source) ¶

overrides orionis.database.contracts.schema.ISchema.drop

Drop an existing table.

Parameters
name:`str`	The name of the table to drop. If the table belongs to a non-default schema, use the `schema.table` format.

Returns
`bool`	`True` when the table is dropped without errors.

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

overrides orionis.database.contracts.schema.ISchema.__slots__

Undocumented

__conn_manager: IConnectionManager = (source) ¶

Undocumented

__connection_name: str | None = (source) ¶

Undocumented

__connection_selected: bool = (source) ¶

Undocumented