sqlalchemy_helpers.aio module

Database management (async).

This must remain independent from any web framework.

class sqlalchemy_helpers.aio.AsyncDatabaseManager(uri: str, alembic_location: str, *, engine_args: MutableMapping[str, Any] | None = None, base_model: type[DeclarativeBase] | None = None)

Bases: BaseDatabaseManager

Helper for a SQLAlchemy and Alembic-powered database, asynchronous version.

Parameters:

• uri – the database URI

• alembic_location – a path to the alembic directory

• engine_args – additional arguments passed to create_async_engine

alembic_cfg

the Alembic configuration object

Type:

alembic.config.Config

engine

the SQLAlchemy Engine instance

Type:

sqlalchemy.engine.Engine

Session

the SQLAlchemy scoped session factory

Type:

sqlalchemy.orm.scoped_session

configured_connection(f: Callable[[Connection], R]) → Callable[[Connection], R]

async create() → None

Create the database tables.

async drop() → None

Drop all the database tables.

async get_current_revision(session: AsyncSession) → str | None

Get the current alembic database revision.

async get_status() → DatabaseStatus

Get the status of the database.

Returns:

the status of the database, see DatabaseStatus.

async sync() → SyncResult

Create or update the database schema.

Returns:

the result of the sync, see SyncResult.

async upgrade(target: str = 'head') → None

Upgrade the database schema.

class sqlalchemy_helpers.aio.Base(**kwargs: Any)

Bases: AsyncAttrs, DeclarativeBase

SQLAlchemy’s base class for async models.

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[registry] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

async sqlalchemy_helpers.aio.get_by_pk(pk: Any, *, session: AsyncSession, model: type[M]) → M | None

Get a model instance using its primary key.

Example

user = get_by_pk(42, session=session, model=User)

async sqlalchemy_helpers.aio.get_one(session: AsyncSession, model: type[M], **attrs: Any) → M

Get an object from the datbase.

Parameters:

• session – The SQLAlchemy session to use

• model – The SQLAlchemy model to query

Returns:

the model instance

async sqlalchemy_helpers.aio.get_or_create(session: AsyncSession, model: type[M], **attrs: Any) → tuple[M, bool]

Function like Django’s get_or_create() method.

It will return a tuple, the first argument being the instance and the second being a boolean: True if the instance has been created and False otherwise.

Example:

user, created = get_or_create(session, User, name="foo")

async sqlalchemy_helpers.aio.update_or_create(session: AsyncSession, model: type[M], defaults: Mapping[str, Any] | None = None, create_defaults: Mapping[str, Any] | None = None, **filter_attrs: Any) → tuple[M, bool]

Function like Django’s update_or_create() method.

It will return a tuple, the first argument being the instance and the second being a boolean: True if the instance has been created and False otherwise.

Example:

user, created = update_or_create(session, User, name="foo", defaults={"full_name": "Foo"})