magpie.db

Attributes

LOGGER

Functions

get_db_url(→ magpie.typedefs.Str)	Retrieve the database connection URL with provided settings.
get_engine(→ sqlalchemy.engine.base.Engine)
get_session_factory(engine)	Create a new session with integrated thread-local scope for safe handling across application workers.
get_tm_session(session_factory, transaction_manager)	Get a sqlalchemy.orm.Session instance backed by a transaction.
get_session_from_other(db_session)
get_db_session_from_settings(...)
get_db_session_from_config_ini(config_ini_path[, ...])
get_connected_session(→ sqlalchemy.orm.session.Session)	Retrieve the session attached to the request or recreated it to ensure it is open and within scoped transaction.
run_database_migration(→ None)	Runs database migration operations with alembic, using the provided session or a new engine connection.
get_database_revision(→ magpie.typedefs.Str)	Obtains the database revision number employed by alembic for schema migration.
is_database_ready(→ bool)	Obtains the database status against expected table names to ensure it is ready for use.
run_database_migration_when_ready(→ None)	Runs db migration if requested by config and need from revisions.
set_sqlalchemy_log_level(→ magpie.typedefs.SettingsType)	Suppresses sqlalchemy verbose logging if not in logging.DEBUG for Magpie.
includeme(config)

Module Contents

magpie.db.LOGGER

magpie.db.get_db_url(username: magpie.typedefs.Str | None = None, password: magpie.typedefs.Str | None = None, db_host: magpie.typedefs.Str | None = None, db_port: magpie.typedefs.Str | int | None = None, db_name: magpie.typedefs.Str | None = None, settings: magpie.typedefs.AnySettingsContainer = None) → magpie.typedefs.Str

Retrieve the database connection URL with provided settings.

magpie.db.get_engine(container: magpie.typedefs.AnySettingsContainer | None = None, prefix: magpie.typedefs.Str = 'sqlalchemy.', **kwargs: Any) → sqlalchemy.engine.base.Engine

magpie.db.get_session_factory(engine)

Create a new session with integrated thread-local scope for safe handling across application workers.

magpie.db.get_tm_session(session_factory, transaction_manager)

Get a sqlalchemy.orm.Session instance backed by a transaction.

This function will hook the session to the transaction manager which will take care of committing any changes.

• When using pyramid_tm it will automatically be committed or aborted depending on whether an exception is raised.

• When using scripts you should wrap the session in a manager yourself. For example:

  import transaction
  
  engine = get_engine(settings)
  session_factory = get_session_factory(engine)
  with transaction.manager:
      db_session = get_tm_session(session_factory, transaction.manager)
  

magpie.db.get_session_from_other(db_session)

magpie.db.get_db_session_from_settings(settings: magpie.typedefs.AnySettingsContainer | None = None, **kwargs: Any) → sqlalchemy.orm.session.Session

magpie.db.get_db_session_from_config_ini(config_ini_path, ini_main_section_name='app:magpie_app', settings_override=None)

magpie.db.get_connected_session(request: pyramid.request.Request) → sqlalchemy.orm.session.Session

Retrieve the session attached to the request or recreated it to ensure it is open and within scoped transaction.

magpie.db.run_database_migration(container: magpie.typedefs.AnySettingsContainer | None = None, db_session: sqlalchemy.orm.session.Session | None = None) → None

Runs database migration operations with alembic, using the provided session or a new engine connection.

magpie.db.get_database_revision(db_session: sqlalchemy.orm.session.Session) → magpie.typedefs.Str

Obtains the database revision number employed by alembic for schema migration.

magpie.db.is_database_ready(db_session: sqlalchemy.orm.session.Session | None = None, container: magpie.typedefs.AnySettingsContainer | None = None) → bool

Obtains the database status against expected table names to ensure it is ready for use.

magpie.db.run_database_migration_when_ready(settings: magpie.typedefs.SettingsType, db_session: sqlalchemy.orm.session.Session | None = None) → None

Runs db migration if requested by config and need from revisions.

magpie.db.set_sqlalchemy_log_level(magpie_log_level: magpie.typedefs.Str | int) → magpie.typedefs.SettingsType

Suppresses sqlalchemy verbose logging if not in logging.DEBUG for Magpie.

magpie.db.includeme(config)