orionis.database.ConnectionManager

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

Constructor: ConnectionManager(app)

View In Hierarchy

Undocumented

Method	`__init__`	Initialize the manager from the application configuration.
Method	`__unknownConnectionMessage`	Build the descriptive message for an unresolved connection name.
Method	`addConnection`	Register or replace a named connection configuration at runtime.
Method	`configFor`	Retrieve the configuration for a named connection.
Method	`connection`	Resolve the connection registered under the given name.
Async Method	`disconnect`	Dispose one or all cached connections.
Method	`getDefaultName`	Return the name of the default connection.
Method	`hasConnection`	Report whether a connection configuration exists.
Method	`setDefaultName`	Change the default connection name.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_cached_connections`	Undocumented
Instance Variable	`_connections`	Undocumented
Instance Variable	`_default`	Undocumented

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

Initialize the manager from the application configuration.

Parameters
app:`IApplication`	Application container providing the database configuration.

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

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

Build the descriptive message for an unresolved connection name.

Parameters
name:`str`	Connection name that failed to resolve.

Returns
`str`	Descriptive message listing the declared connection names.

def addConnection(self, name: str, config: dict[str, Any]): (source) ¶

overrides orionis.database.contracts.IConnectionManager.addConnection

A live connection built from a previous configuration keeps serving until disconnect is called for that name.

Parameters
name:`str`	Connection name to register.
config:`dict`	Driver configuration for the connection.

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

Raises
`ValueError`	If the name is empty.
`TypeError`	If the configuration is not a mapping.

def configFor(self, name: str | None = None) -> dict[str, Any]: (source) ¶

overrides orionis.database.contracts.IConnectionManager.configFor

Retrieve the configuration for a named connection.

Parameters
name:`str or None, optional`	Connection name to look up, or `None` for the default.

Returns
`dict[str, Any]`	The configuration dictionary for the requested connection.

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

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

overrides orionis.database.contracts.IConnectionManager.connection

Resolve the connection registered under the given name.

The connection is built on first access and cached for reuse.

Parameters
name:`str or None, optional`	Connection name as declared in the database configuration, or `None` for the default connection.

Returns
`IConnection`	Connection bound to its configuration.

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

async def disconnect(self, name: str | None = None): (source) ¶

overrides orionis.database.contracts.IConnectionManager.disconnect

Dispose one or all cached connections.

Parameters
name:`str or None, optional`	Connection to dispose, or `None` to dispose every cached connection.

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

def getDefaultName(self) -> str: (source) ¶

overrides orionis.database.contracts.IConnectionManager.getDefaultName

Return the name of the default connection.

Returns
`str`	Default connection name as configured.

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

overrides orionis.database.contracts.IConnectionManager.hasConnection

Report whether a connection configuration exists.

Parameters
name:`str`	Connection name to look up.

Returns
`bool`	`True` when the configuration is registered.

def setDefaultName(self, name: str): (source) ¶

overrides orionis.database.contracts.IConnectionManager.setDefaultName

Change the default connection name.

Parameters
name:`str`	Connection name to use as the default; must already be declared in the configuration.

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

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

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

overrides orionis.database.contracts.IConnectionManager.__slots__

Undocumented

_cached_connections: dict[str, IConnection] = (source) ¶

Undocumented

_connections: dict[str, dict[str, Any]] = (source) ¶

Undocumented

_default: str = (source) ¶

Undocumented