babylon.engine.database

Injectable database connection for the simulation engine.

This module provides a DatabaseConnection class that wraps SQLAlchemy engine and session creation, enabling dependency injection for testing.

Unlike the module-level singletons in babylon.data.database, this class allows creating isolated database connections for each test or component.

Sprint 3: Central Committee (Dependency Injection)

Classes

DatabaseConnection([url])

Injectable database connection wrapper.

class babylon.engine.database.DatabaseConnection(url='sqlite:///babylon.db')[source]

Bases: object

Injectable database connection wrapper.

Wraps SQLAlchemy engine and sessionmaker to provide clean resource management and testability.

Example

>>> db = DatabaseConnection(url="sqlite:///:memory:")
>>> with db.session() as session:
...     result = session.execute(text("SELECT 1"))
...     print(result.scalar())
1
>>> db.close()
Parameters:

url (str)

__init__(url='sqlite:///babylon.db')[source]

Initialize database connection.

Parameters:

url (str) – SQLAlchemy database URL. Defaults to local SQLite file. Use “sqlite:///:memory:” for in-memory testing.

Return type:

None

session()[source]

Get a database session within a context manager.

The session is automatically closed when the context exits. On exception, the session is rolled back.

Yields:

SQLAlchemy Session object

Return type:

Iterator[Session]

Example

>>> with db.session() as session:
...     session.execute(text("INSERT INTO ..."))
...     session.commit()
Return type:

Iterator[Session]

close()[source]

Dispose of the database engine and release resources.

After calling close(), the connection cannot be used. Attempting to create new sessions will fail.

Return type:

None