SessionManager

Manage SQLAlchemy sessions.

Attributes

Attribute	Type	Description
_engines	dict = {}	A dictionary mapping database URIs to their respective SQLAlchemy engine instances for reuse across the application.
_sessions	dict = {}	A dictionary mapping database URIs to sessionmaker factories used to generate new database sessions.
forked	boolean = false	A boolean flag indicating whether the process has been forked, used to determine if engines and sessions should be cached or recreated.
prepared	boolean = false	A boolean flag that tracks whether the database models and metadata have been successfully initialized on the target engine.
engine_callback	callable = null	An optional callable that is executed whenever a new SQLAlchemy engine is created, typically used for engine configuration or logging.

---

Constructor

Signature

def SessionManager(
    engine_callback: callable = None
) - > null

Parameters

Name	Type	Description
engine_callback	callable = None	An optional callback function to be executed whenever a new SQLAlchemy engine is created.

---

Methods

---

get_engine()

@classmethod
def get_engine(
    dburi: string,
    **kwargs: dict
) - > Engine

Retrieves or creates a SQLAlchemy engine for the specified database URI, using connection pooling if forked or a NullPool otherwise.

Parameters

Name	Type	Description
dburi	string	The connection string used to identify and connect to the target database.
**kwargs	dict	Additional configuration arguments passed to the SQLAlchemy create_engine call.

Returns

Type	Description
Engine	The SQLAlchemy engine instance configured for the provided database URI.

---

create_session()

@classmethod
def create_session(
    dburi: string,
    short_lived_sessions: boolean,
    **kwargs: dict
) - > tuple

Initializes a session factory and its associated engine for a given database URI.

Parameters

Name	Type	Description
dburi	string	The connection string for the database where the session will be bound.
short_lived_sessions	boolean	If true, prevents the session factory from being cached in the internal registry.
**kwargs	dict	Configuration parameters used when initializing the database engine.

Returns

Type	Description
tuple	A tuple containing the SQLAlchemy engine and the sessionmaker factory.

---

invalidate()

@classmethod
def invalidate(
    dburi: string
)

Dispose cached engine/session state for a database URI.

Parameters

Name	Type	Description
dburi	string	The database connection string whose associated engine and session factory should be cleared and disposed.

---

prepare_models()

@classmethod
def prepare_models(
    engine: Engine
)

Ensures all database tables defined in the model metadata exist, implementing a retry mechanism to handle race conditions during creation.

Parameters

Name	Type	Description
engine	Engine	The SQLAlchemy engine used to execute the DDL commands for table creation.

---

session_factory()

@classmethod
def session_factory(
    dburi: string,
    **kwargs: dict
) - > Session

Creates a new database session after ensuring that the necessary database models have been prepared.

Parameters

Name	Type	Description
dburi	string	The connection string for the database to session into.
**kwargs	dict	Configuration options passed through to the engine and session creation logic.

Returns

Type	Description
Session	A new SQLAlchemy Session instance ready for database operations.

---