OCDocker.DB.DBMinimal module¶

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

Usage:

import OCDocker.DB.DBMinimal as ocdbmin

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)[source]¶

Create the database if it does not exist.

Parameters:: url (str | sqlalchemy.engine.url.URL) – The database url (string or URL object).
Return type:: None

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