sqlalchemy_helpers.manager module

Database management.

This must remain independent from any web framework.

sqlalchemy_helpers.manager.Base

SQLAlchemy’s base class for models.

Type:

object

class sqlalchemy_helpers.manager.DatabaseManager(uri, alembic_location, engine_args=None, base_model=None)

Bases: object

Helper for a SQLAlchemy and Alembic-powered database

Parameters:

• uri (str) – the database URI

• alembic_location (str) – a path to the alembic directory

• engine_args (dict) – 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()

Create the database tables.

drop()

Drop all the database tables.

get_current_revision(session)

Get the current alembic database revision.

get_latest_revision()

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

get_status()

Get the status of the database.

Returns:

see DatabaseStatus.

Return type:

DatabaseStatus member

sync()

Create or update the database schema.

Returns:

see SyncResult.

Return type:

SyncResult member

upgrade(target='head')

Upgrade the database schema.

class sqlalchemy_helpers.manager.DatabaseStatus(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

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(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

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, tablename, columnname=None)

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

Parameters:

• bind (sqlalchemy.engine.Engine) – the database engine or connection.

• tablename (str) – the table to look for.

• columnname (str, optional) – the column to look for, if any. Defaults to None.

Returns:

Whether the database (and column) exist.

Return type:

bool

sqlalchemy_helpers.manager.get_base(*args, **kwargs)

A wrapper for declarative_base().

sqlalchemy_helpers.manager.get_by_pk(pk, *, session, model)

Get a model instance using its primary key.

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

sqlalchemy_helpers.manager.get_one(session, model, **attrs)

Get a model instance using filters.

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

sqlalchemy_helpers.manager.get_or_create(session, model, **attrs)

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)

Check whether the database is SQLite.

Returns:

whether the database is SQLite.

Return type:

bool

sqlalchemy_helpers.manager.model_property(func)

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

sqlalchemy_helpers.manager.session_and_model_property(Session, func)

Add a model property that uses the database session.

sqlalchemy_helpers.manager.set_sqlite_pragma(dbapi_connection, connection_record)

Automatically activate foreign keys on SQLite databases.