sqlalchemy_helpers.manager module

Database management.

This must remain independent from any web framework.

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

Bases: DeclarativeBase

SQLAlchemy’s base class for 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.

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

Bases: object

Helper for a SQLAlchemy and Alembic-powered database

Parameters:

• uri – the database URI

• alembic_location – a path to the alembic directory

• engine_args – additional arguments passed to create_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

get_latest_revision() → str | None

Get the most up-to-date alembic database revision available.

class sqlalchemy_helpers.manager.DatabaseManager(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

Parameters:

• uri – the database URI

• alembic_location – a path to the alembic directory

• engine_args – additional arguments passed to create_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

create() → None

Create the database tables.

drop() → None

Drop all the database tables.

get_current_revision(session: Session | None = None) → str | None

Get the current alembic database revision.

Parameters:

session – the session instance to use, or None if one is to be created.

get_status(session: Session | None = None) → DatabaseStatus

Get the status of the database.

Parameters:

session – the session instance to use, or None if one is to be created.

Returns:

the database status, see DatabaseStatus.

sync(session: Session | None = None) → SyncResult

Create or update the database schema.

Parameters:

session – the session instance to use, or None if one is to be created.

Returns:

the result of the sync, see SyncResult.

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

Upgrade the database schema.

class sqlalchemy_helpers.manager.DatabaseStatus(*values)

Bases: Enum

The status of the database.

NO_INFO = 2

Returned when the database couldn’t be connected to.

UPGRADE_AVAILABLE = 3

Returned when the database schema can be upgraded.

UP_TO_DATE = 1

Returned when the database schema is up-to-date.

class sqlalchemy_helpers.manager.SyncResult(*values)

Bases: Enum

The result of a sync() call.

ALREADY_UP_TO_DATE = 1

Returned when the database schema was already up-to-date.

CREATED = 2

Returned when the database has been created.

UPGRADED = 3

Returned when the database schema has been upgraded.

sqlalchemy_helpers.manager.exists_in_db(bind: Engine | Connection, tablename: str, columnname: str | None = None) → bool

Check whether a table and optionally a column exist in the database.

Parameters:

• bind – the database engine or connection.

• tablename – the table to look for.

• columnname – the column to look for, if any. Defaults to None.

Returns:

Whether the database (and column) exist.

sqlalchemy_helpers.manager.get_base(*args: Any, **kwargs: Any) → type[DeclarativeBase]

sqlalchemy_helpers.manager.get_by_pk(pk: Any, *, session: Session, model: type[M]) → M | None

Get a model instance using its primary key.

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

sqlalchemy_helpers.manager.get_one(session: Session, model: type[M], **attrs: Any) → M

Get a model instance using filters.

Example: user = get_one(session, User, name="foo")

sqlalchemy_helpers.manager.get_or_create(session: Session, 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")

sqlalchemy_helpers.manager.is_sqlite(bind: Engine | Connection) → bool

Check whether the database is SQLite.

Returns:

whether the database is SQLite.

Return type:

bool

sqlalchemy_helpers.manager.model_property(func: Callable[[...], Any]) → Any

Add a model property to call a function that uses the database model.

sqlalchemy_helpers.manager.session_and_model_property(Session: scoped_session[Session], func: Callable[[...], Any]) → Any

Add a model property that uses the database session.

sqlalchemy_helpers.manager.set_sqlite_pragma(dbapi_connection: Any, connection_record: Any) → None

Automatically activate foreign keys on SQLite databases.

sqlalchemy_helpers.manager.update_or_create(session: Session, 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"})