orionis.database.dialect

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

module documentation

(source)

Undocumented

Class	`_MySQLParameterEscaper`	Escape binary parameters independently of aiomysql's removed converter.
Function	`_config_text`	Extract a trimmed text value from a configuration mapping.
Function	`_configure_mysql_binary_parameters`	Install safe binary conversion on one adapted aiomysql connection.
Function	`_enum_value`	Extract the primitive value from a possible enum member.
Function	`_is_sqlite_memory`	Report whether a SQLite configuration targets an in-memory database.
Function	`_mysql_session_commands`	Build the MySQL session commands for a configuration.
Function	`_normalize_switch`	Normalize a boolean-like configuration value to `ON` or `OFF`.
Function	`_oracle_query`	Build the URL query parameters for an Oracle connection.
Function	`_oracle_url`	Build the engine URL for an Oracle connection.
Function	`_pgsql_connect_args`	Build the asyncpg connect arguments for a PostgreSQL connection.
Function	`_server_query`	Build the URL query parameters for host-based drivers.
Function	`_server_url`	Build the engine URL for host-based drivers.
Function	`_session_statements`	Build the per-connection session statements for a configuration.
Function	`_sqlite_pragmas`	Build the PRAGMA statements for a SQLite configuration.
Function	`_sqlite_url`	Build the engine URL for a SQLite connection.
Function	`_yes_no`	Normalize a boolean-like configuration value to `yes` or `no`.
Function	`build_engine_url`	Build the engine URL for a connection configuration.
Function	`configure_engine`	Apply driver-specific session settings to a freshly built engine.
Function	`engine_options`	Build keyword options for an engine factory.
Function	`missing_dependency_error`	Build the exception raised when a DB driver package is absent.
Function	`resolve_driver`	Extract and validate the driver name from a connection configuration.
Constant	`_ASYNC_DIALECTS`	Undocumented
Constant	`_ASYNC_DRIVER_PACKAGES`	Undocumented
Constant	`_DEFAULT_ODBC_DRIVER`	Undocumented
Constant	`_IDENTIFIER_PATTERN`	Undocumented
Constant	`_MYSQL_RELAXED_MODE`	Undocumented
Constant	`_MYSQL_STRICT_MODE`	Undocumented
Constant	`_SQLITE_MEMORY_MARKERS`	Undocumented
Constant	`_SYNC_DIALECTS`	Undocumented
Constant	`_SYNC_DRIVER_PACKAGES`	Undocumented

def _config_text(config: dict[str, Any], key: str) -> str | None: (source) ¶

Extract a trimmed text value from a configuration mapping.

Parameters
config:`dict`	Connection configuration.
key:`str`	Configuration key to read.

Returns
`str or None`	Trimmed value, or `None` when empty or absent.

def _configure_mysql_binary_parameters(dbapi_connection: object): (source) ¶

Install safe binary conversion on one adapted aiomysql connection.

PyMySQL 1.2.3 keeps escape_bytes_prefixed as a noncallable import alias; aiomysql 0.3.2 calls that alias for every bytes parameter. Hex literals keep binary storage working without downgrading PyMySQL or patching its modules.

Parameters
dbapi_connection:`object`	SQLAlchemy DBAPI adapter exposing its underlying driver connection.

Returns
`None`	Replace only this connection's binary escape path when available.

def _enum_value(value: Any) -> str: (source) ¶

Extract the primitive value from a possible enum member.

Parameters
value:`Any`	Enum member, string, or `None`.

Returns
`str`	String form of the value; empty when the input is `None`.

def _is_sqlite_memory(config: dict[str, Any]) -> bool: (source) ¶

Report whether a SQLite configuration targets an in-memory database.

Parameters
config:`dict`	SQLite connection configuration.

Returns
`bool`	`True` for in-memory databases.

def _mysql_session_commands(config: dict[str, Any]) -> tuple[str, ...]: (source) ¶

Build the MySQL session commands for a configuration.

Applies the connection charset and collation through SET NAMES and the strict (or relaxed) sql_mode preset, mirroring the behavior of mainstream frameworks.

Parameters
config:`dict`	MySQL connection configuration.

Returns
`tuple of str`	Session commands to run on each new pooled connection.

def _normalize_switch(value: Any) -> str: (source) ¶

Normalize a boolean-like configuration value to ON or OFF.

Parameters
value:`Any`	Boolean, string, or enum member describing the switch state.

Returns
`str`	`"ON"` or `"OFF"`.

def _oracle_query(config: dict[str, Any], sid: str | None) -> dict[str, str]: (source) ¶

Build the URL query parameters for an Oracle connection.

Parameters
config:`dict`	Oracle connection configuration.
sid:`str or None`	Resolved SID; service-name addressing applies only without it.

Returns
`dict of str to str`	Query parameters for the engine URL.

def _oracle_url(config: dict[str, Any], dialect: str) -> URL: (source) ¶

Build the engine URL for an Oracle connection.

Parameters
config:`dict`	Oracle connection configuration.
dialect:`str`	SQLAlchemy dialect name to build the URL for.

Returns
`URL`	Oracle engine URL using service name or SID addressing.

def _pgsql_connect_args(config: dict[str, Any]) -> dict[str, Any]: (source) ¶

Build the asyncpg connect arguments for a PostgreSQL connection.

Maps sslmode to the driver ssl argument and forwards the configured search_path and charset as server settings.

Parameters
config:`dict`	PostgreSQL connection configuration.

Returns
`dict`	Driver connect arguments; empty when nothing is configured.

def _server_query(driver: str, config: dict[str, Any]) -> dict[str, str]: (source) ¶

Build the URL query parameters for host-based drivers.

Parameters
driver:`str`	Normalized driver name.
config:`dict`	Connection configuration.

Returns
`dict of str to str`	Query parameters for the engine URL.

def _server_url(driver: str, config: dict[str, Any], dialect: str) -> URL: (source) ¶

Build the engine URL for host-based drivers.

Covers MySQL, PostgreSQL, and SQL Server connections addressed by host, port, and database name.

Parameters
driver:`str`	Normalized driver name.
config:`dict`	Connection configuration.
dialect:`str`	SQLAlchemy dialect name to build the URL for.

Returns
`URL`	Engine URL with credentials, host, port, and database.

def _session_statements(config: dict[str, Any]) -> tuple[str, ...]: (source) ¶

Build the per-connection session statements for a configuration.

Parameters
config:`dict`	Connection configuration.

Returns
`tuple of str`	Statements to run on each new pooled connection.

def _sqlite_pragmas(config: dict[str, Any]) -> tuple[str, ...]: (source) ¶

Build the PRAGMA statements for a SQLite configuration.

Parameters
config:`dict`	SQLite connection configuration.

Returns
`tuple of str`	PRAGMA statements to run on each new connection.

def _sqlite_url(config: dict[str, Any], dialect: str) -> URL: (source) ¶

Build the engine URL for a SQLite connection.

The engine URL is always derived from the database path; the informational url key (sync-style DSN) is intentionally ignored.

Parameters
config:`dict`	SQLite connection configuration.
dialect:`str`	SQLAlchemy dialect name to build the URL for.

Returns
`URL`	SQLite engine URL.

def _yes_no(value: Any) -> str: (source) ¶

Normalize a boolean-like configuration value to yes or no.

Parameters
value:`Any`	Boolean, string, or enum member describing the switch state.

Returns
`str`	`"yes"` or `"no"`.

def build_engine_url(config: dict[str, Any], *, sync: bool = False) -> URL: (source) ¶

Build the engine URL for a connection configuration.

Parameters
config:`dict`	Connection configuration produced by the database config entities.
sync:`bool, optional`	Whether to select the blocking DBAPI dialect instead of the async one.

Returns
`URL`	Engine URL for the configured driver.

Raises
`UnsupportedDriverException`	If the driver has no registered dialect.

def configure_engine(engine: AsyncEngine, config: dict[str, Any]): (source) ¶

Apply driver-specific session settings to a freshly built engine.

For SQLite this installs a connect hook applying the configured PRAGMA settings; for MySQL it applies the connection charset, collation, and strict mode on every new pooled connection.

Parameters
engine:`AsyncEngine`	Engine to configure.
config:`dict`	Connection configuration.

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

def engine_options(config: dict[str, Any], *, sync: bool = False) -> dict[str, Any]: (source) ¶

Build keyword options for an engine factory.

Parameters
config:`dict`	Connection configuration.
sync:`bool, optional`	Whether to select the blocking DBAPI driver options instead of the async ones.

Returns
`dict`	Options such as pool class and driver connect arguments.

def missing_dependency_error(driver: str, cause: ModuleNotFoundError, *, sync: bool = False) -> MissingDatabaseDependencyException: (source) ¶

Build the exception raised when a DB driver package is absent.

Parameters
driver:`str`	Orionis driver name whose package is missing.
cause:`ModuleNotFoundError`	Original import error raised by the engine.
sync:`bool, optional`	Whether the missing package is the blocking (synchronous) driver instead of the default async one.

Returns
`MissingDatabaseDependencyException`	Exception with an actionable uv installation hint.

def resolve_driver(config: dict[str, Any]) -> str: (source) ¶

Extract and validate the driver name from a connection configuration.

Parameters
config:`dict`	Connection configuration containing a `driver` key.

Returns
`str`	Normalized driver name.

Raises
`UnsupportedDriverException`	If the driver is missing or has no registered dialect.

_ASYNC_DIALECTS: dict[str, str] = (source) ¶

Undocumented

Value

{'sqlite': 'sqlite+aiosqlite',
 'mysql': 'mysql+aiomysql',
 'pgsql': 'postgresql+asyncpg',
 'oracle': 'oracle+oracledb_async',
 'sqlserver': 'mssql+aioodbc'}

_ASYNC_DRIVER_PACKAGES: dict[str, tuple[str, str]] = (source) ¶

Undocumented

Value

{'sqlite': ('aiosqlite', 'orionis'),
 'mysql': ('aiomysql', 'orionis[mysql]'),
 'pgsql': ('asyncpg', 'orionis[pgsql]'),
 'oracle': ('oracledb', 'orionis[oracle]'),
 'sqlserver': ('aioodbc', 'orionis[sqlserver]')}

_DEFAULT_ODBC_DRIVER: str = (source) ¶

Undocumented

Value

'ODBC Driver 18 for SQL Server'

_IDENTIFIER_PATTERN: re.Pattern[str] = (source) ¶

Undocumented

Value

re.compile(r'^\w+$',
           re.ASCII)

_MYSQL_RELAXED_MODE: str = (source) ¶

Undocumented

Value

'NO_ENGINE_SUBSTITUTION'

_MYSQL_STRICT_MODE: str = (source) ¶

Undocumented

Value

'ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_D↵
IVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION'

_SQLITE_MEMORY_MARKERS: frozenset[str] = (source) ¶

Undocumented

Value

frozenset(set([':memory:', '']))

_SYNC_DIALECTS: dict[str, str] = (source) ¶

Undocumented

Value

{'sqlite': 'sqlite',
 'mysql': 'mysql+pymysql',
 'pgsql': 'postgresql+psycopg2',
 'oracle': 'oracle+oracledb',
 'sqlserver': 'mssql+pyodbc'}

_SYNC_DRIVER_PACKAGES: dict[str, tuple[str, str]] = (source) ¶

Undocumented

Value

{'sqlite': ('sqlite3', 'the Python standard library'),
 'mysql': ('pymysql', 'orionis[mysql]'),
 'pgsql': ('psycopg2', 'orionis[pgsql]'),
 'oracle': ('oracledb', 'orionis[oracle]'),
 'sqlserver': ('pyodbc', 'orionis[sqlserver]')}