API reference

Components

class asphalt.sqlalchemy.component.SQLAlchemyComponent(*, url=None, bind=None, prefer_async=True, engine_args=None, session_args=None, commit_executor_workers=5, ready_callback=None, poolclass=None, resource_name='default')

Creates resources necessary for accessing relational databases using SQLAlchemy.

This component supports both synchronous (sqlite, psycopg2, etc.) and asynchronous (asyncpg, asyncmy, etc.) engines, and the provided resources differ based on that.

For synchronous engines, the following resources are provided:

For asynchronous engines, the following resources are provided:

Note

The following options will always be set to fixed values in sessions:

expire_on_commit: False
future: True

Parameters:

url (str | URL | dict[str, Any] | None) – the connection url passed to create_engine() (can also be a dictionary of URL keyword arguments)
bind (Connection | Engine | AsyncConnection | AsyncEngine | None) – a connection or engine to use instead of creating a new engine
prefer_async (bool) – if True, try to create an async engine rather than a synchronous one, in cases like psycopg where the driver supports both
engine_args (dict[str, Any] | None) – extra keyword arguments passed to sqlalchemy.future.engine.create_engine() or sqlalchemy.ext.asyncio.create_engine()
session_args (dict[str, Any] | None) – extra keyword arguments passed to Session or AsyncSession
commit_executor_workers (int) – maximum number of worker threads to use for tearing down synchronous sessions (default: 5; ignored for asynchronous engines)
ready_callback (Callable[[Engine, sessionmaker], Any] | str | None) – a callable that is called right before the resources are added to the context (can be a coroutine function too)
poolclass (str | type[Pool] | None) – the SQLAlchemy pool class (or a textual reference to one) to use; passed to sqlalchemy.future.engine.create_engine() or sqlalchemy.ext.asyncio.create_engine()
resource_name (str) – name space for the database resources

Utilities

asphalt.sqlalchemy.utils.clear_database(engine, schemas=())

Clear any tables from an existing database using a synchronous connection/engine.

Parameters:

engine (Engine | Connection) – the engine or connection to use
schemas (Iterable[str]) – full list of schema names to expect (ignored for SQLite)

Return type:

None

async asphalt.sqlalchemy.utils.clear_async_database(connection, schemas=())

Clear any tables from an existing database using an asynchronous connection.

Parameters:

connection – the connection to use
schemas – full list of schema names to expect (ignored for SQLite)

asphalt.sqlalchemy.utils.apply_sqlite_hacks(engine)

Apply hacks for SAVEPOINT support on pysqlite based engines.

This function is automatically called by the component, and only needs to be explicitly used by the developer when using an SQLite connection for database integration tests (the connection is passed to the component as the bind option).