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:
Engine
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 ofURL
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 likepsycopg
where the driver supports bothengine_args (dict[str, Any] | None) – extra keyword arguments passed to
sqlalchemy.future.engine.create_engine()
orsqlalchemy.ext.asyncio.create_engine()
session_args (dict[str, Any] | None) – extra keyword arguments passed to
Session
orAsyncSession
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()
orsqlalchemy.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).- Parameters:
engine (Engine | AsyncEngine) – an engine using the sqlite dialect
- Return type:
None