OCDocker.DB.DBMinimal module¶

Sets of classes and functions that are used for setting up the database.

Usage:

import OCDocker.DB.DBMinimal as ocdbmin

exception OCDocker.DB.DBMinimal.DatabaseError[source]¶

Bases: RuntimeError

Base exception for database setup and access failures.

exception OCDocker.DB.DBMinimal.MissingDatabaseDependencyError[source]¶

Bases: DatabaseError, ImportError

Raised when an optional database dependency is not installed.

exception OCDocker.DB.DBMinimal.DatabaseConfigurationError[source]¶

Bases: DatabaseError, ValueError

Raised when database configuration is invalid or incomplete.

exception OCDocker.DB.DBMinimal.DatabaseCreationNotAllowedError[source]¶

Bases: DatabaseError

Raised when a missing remote database would be created implicitly.

exception OCDocker.DB.DBMinimal.DatabaseCreationError[source]¶

Bases: DatabaseError

Raised when explicit database creation fails.

exception OCDocker.DB.DBMinimal.DatabaseConnectionError[source]¶

Bases: DatabaseError

Raised when a database connection/check fails.

class OCDocker.DB.DBMinimal.DatabaseSettings(backend, host='', user='', password='', database='', optimizedb='', port=None, sqlite_path='')[source]¶

Bases: object

Validated database connection settings.

Parameters:

backend (str)
host (str)
user (str)
password (str)
database (str)
optimizedb (str)
port (int | None)
sqlite_path (str)

backend: str¶

host: str = ''¶

user: str = ''¶

password: str = ''¶

database: str = ''¶

optimizedb: str = ''¶

port: int | None = None¶

sqlite_path: str = ''¶

OCDocker.DB.DBMinimal.normalize_db_backend(raw_backend)[source]¶

Normalize backend names and SQLAlchemy driver names.

Parameters:: raw_backend (str)
Return type:: str | None

OCDocker.DB.DBMinimal.database_exists(url)[source]¶

Return whether a database exists using sqlalchemy-utils.

Parameters:: url (str | URL)
Return type:: bool

OCDocker.DB.DBMinimal.create_database(url)[source]¶

Create a database using sqlalchemy-utils.

Parameters:: url (str | URL)
Return type:: None

OCDocker.DB.DBMinimal.validate_database_config(backend, *, host='', user='', password='', database='', optimizedb='', port=None, sqlite_path='', require_credentials=False, require_sqlite_path=False)[source]¶

Validate database connection configuration.

Parameters:

backend (str) – Database backend name or alias.
require_credentials (bool, optional) – Require host/user/password/database for PostgreSQL/MySQL.
require_sqlite_path (bool, optional) – Require an explicit SQLite path instead of using :memory:.
host (str)
user (str)
password (str)
database (str)
optimizedb (str)
port (int | str | None)
sqlite_path (str)

Return type:

DatabaseSettings

OCDocker.DB.DBMinimal.build_database_urls(settings)[source]¶

Build primary and optimization SQLAlchemy URLs from validated settings.

Parameters:: settings (DatabaseSettings)
Return type:: Tuple[URL, URL]

OCDocker.DB.DBMinimal.get_default_engine()[source]¶

Return the explicitly initialized default engine, if present.

Return type:: Any

OCDocker.DB.DBMinimal.get_default_session()[source]¶

Return the explicitly initialized default session factory, if present.

Return type:: Any

OCDocker.DB.DBMinimal.cleanup_engine(engine)[source]¶

Clean up an engine by disposing of all connections in the pool.

This function closes all connections in the connection pool and disposes of the engine. It’s automatically called on application shutdown via atexit handlers.

Parameters:: engine (Engine | None) – The engine to clean up.
Return type:: None

Notes

This is safe to call multiple times (idempotent)
Errors during cleanup are silently ignored
Typically called automatically on application exit
Prevents connection leaks, especially important for pooled DB backends

OCDocker.DB.DBMinimal.cleanup_session(session)[source]¶

Clean up a scoped session by removing all sessions from the registry.

This function removes all thread-local session instances from the scoped_session registry. It’s automatically called on application shutdown via atexit handlers.

Parameters:: session (scoped_session | None) – The scoped session to clean up.
Return type:: None

Notes

This is safe to call multiple times (idempotent)
Errors during cleanup are silently ignored
Typically called automatically on application exit

OCDocker.DB.DBMinimal.create_database_if_not_exists(url, *, create_if_missing=False)[source]¶

Create the database only when explicitly allowed.

Parameters:

url (str | sqlalchemy.engine.url.URL) – The database url (string or URL object).
create_if_missing (bool, optional) – If True, create a missing database. The default is False to avoid silently creating remote PostgreSQL/MySQL databases.

Returns:

True if this call created a database, False otherwise.

Return type:

bool

OCDocker.DB.DBMinimal.create_engine(url, echo=False, pool_size=5, max_overflow=10, pool_timeout=30, pool_recycle=3600)[source]¶

Create the engine with connection pooling.

Parameters:

url (str | sqlalchemy.engine.url.URL) – The database url (string or URL object).
echo (bool) – Echo the SQL commands.
pool_size (int, optional) – Number of connections to maintain in the pool. Default is 5.
max_overflow (int, optional) – Maximum number of connections to allow beyond pool_size. Default is 10.
pool_timeout (int, optional) – Seconds to wait before giving up on getting a connection. Default is 30.
pool_recycle (int, optional) – Seconds after which a connection is recreated. Default is 3600 (1 hour).

Returns:

Engine – The engine with connection pooling configured.

Return type:

sqlalchemy.engine.base.Engine

OCDocker.DB.DBMinimal.create_session(engine)[source]¶

Create a scoped session for database operations.

Parameters:: engine (from sqlalchemy.engine.base.Engine | None) – The engine.
Returns:: scoped_session – The scoped session factory. Use with session() as s: to get a session instance.
Return type:: sqlalchemy.orm.scoped_session

Notes

Session Lifecycle: - Always use context managers: with session() as s: … - The context manager automatically handles commit/rollback and closing - The scoped_session registry is cleaned up automatically on application shutdown - For manual cleanup, call cleanup_session(session) or let atexit handlers run

Example

session = create_session(engine)
with session() as s:
    result = s.query(Model).all()
    s.commit()  # Optional - context manager handles this