Configuration

Single database configuration

You can initialise the manager providing an instance of SQLAlchemyConfig

from sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager

config = SQLAlchemyConfig(
    engine_url="sqlite:///./sqlite.db",
    engine_options=dict(connect_args={"check_same_thread": False}, echo=True),
    session_options=dict(expire_on_commit=False),
)

sa_manager = SQLAlchemyBindManager(config)

The engine_url and engine_options dictionaries accept the same parameters as SQLAlchemy create_engine()

The session_options dictionary accepts the same parameters as SQLALchemy sessionmaker()

Once the bind manager is initialised we can retrieve and use the SQLAlchemyBind using the method get_bind()

The SQLAlchemyBind class has the following attributes:

• engine: The initialised SQLALchemy Engine
• declarative_base: A base class that can be used to create declarative models
• registry_mapper: The registry associated with the engine. It can be used with Alembic or to setup imperative mapping
• session_class: The class built by sessionmaker(), either Session or AsyncSession

The SQLAlchemyBindManager provides some helper methods to quickly access some of the bind properties without using the SQLAlchemyBind:

• get_session: returns a Session object
• get_mapper: returns the mapper associated with the bind

Multiple databases configuration

SQLAlchemyBindManager accepts also multiple databases configuration, provided as a dictionary. The dictionary keys are used as a reference name for each bind.

from sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager

config = {
    "default": SQLAlchemyConfig(
        engine_url="sqlite:///./sqlite.db",
        engine_options=dict(connect_args={"check_same_thread": False}, echo=True),
        session_options=dict(expire_on_commit=False),
    ),
    "secondary": SQLAlchemyConfig(
        engine_url="sqlite:///./secondary.db",
        engine_options=dict(connect_args={"check_same_thread": False}, echo=True),
        session_options=dict(expire_on_commit=False),
    ),
}

sa_manager = SQLAlchemyBindManager(config)

All the SQLAlchemyBindManager helper methods accept the bind_name optional parameter:

• get_session(bind_name="default")
• get_mapper(bind_name="default")

Asynchronous database engines

Is it possible to supply configurations for asyncio supported engines using the async_engine config property.

from sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager

config = SQLAlchemyConfig(
    engine_url="postgresql+asyncpg://scott:tiger@localhost/test",
    async_engine=True,
)

sa_manager = SQLAlchemyBindManager(config)

This will make sure we initialise a SQLAlchemyAsyncBind object, containing AsyncEngine and AsyncSession.

The interfaces are exactly the same, however you'll need to await the relevant functions in SQLAlchemy. E.g.:

async with sa_manager.get_session() as session:
    session.add(o)
    await session.commit()

Note that async implementation has several differences from the sync one, make sure to check SQLAlchemy asyncio documentation

Bind engines lifecycle

Engine disposal is handled automatically by SQLAlchemyBindManager. Engines are disposed when:

• The manager instance is garbage collected
• The Python interpreter shuts down (via an atexit handler)

In some scenarios, such as automated tests, you may need to manually dispose engines to release database connections between tests. The dispose_engines() method is available for this purpose:

sa_manager = SQLAlchemyBindManager(config)

# ... use the manager ...

# Manually dispose all engines
sa_manager.dispose_engines()

This method disposes all engines synchronously, including async engines (using their underlying sync engine).