magpie.db

Module Contents

Functions

get_db_url(username: Optional[Str] = None, password: Optional[Str] = None, db_host: Optional[Str] = None, db_port: Optional[Union[Str, int]] = None, db_name: Optional[Str] = None, settings: AnySettingsContainer = None) → Str	Retrieve the database connection URL with provided settings.
get_engine(container: Optional[AnySettingsContainer] = None, prefix: Str = ‘sqlalchemy.’, **kwargs: Any) → Engine
get_session_factory(engine)
get_tm_session(session_factory, transaction_manager)	Get a sqlalchemy.orm.Session instance backed by a transaction.
get_db_session_from_settings(settings: Optional[AnySettingsContainer] = None, **kwargs: Any) → Session
get_db_session_from_config_ini(config_ini_path, ini_main_section_name=’app:magpie_app’, settings_override=None)
run_database_migration(settings: Optional[SettingsType] = None, db_session: Optional[Session] = None) → None	Runs database migration operations with alembic, using the provided session or a new engine connection.
get_database_revision(db_session: Session) → Str	Obtains the database revision number employed by alembic for schema migration.
is_database_ready(db_session: Optional[Session] = None, container: Optional[AnySettingsContainer] = None) → bool	Obtains the database status against expected table names to ensure it is ready for use.
run_database_migration_when_ready(settings: SettingsType, db_session: Optional[Session] = None) → None	Runs db migration if requested by config and need from revisions.
set_sqlalchemy_log_level(magpie_log_level: Union[Str, int]) → SettingsType	Suppresses sqlalchemy verbose logging if not in logging.DEBUG for Magpie.
includeme(config)

magpie.db.LOGGER

magpie.db.get_db_url(username: Optional[Str] = None, password: Optional[Str] = None, db_host: Optional[Str] = None, db_port: Optional[Union[Str, int]] = None, db_name: Optional[Str] = None, settings: AnySettingsContainer = None) → Str

Retrieve the database connection URL with provided settings.

magpie.db.get_engine(container: Optional[AnySettingsContainer] = None, prefix: Str = 'sqlalchemy.', **kwargs: Any) → Engine

magpie.db.get_session_factory(engine)

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_db_session_from_settings(settings: Optional[AnySettingsContainer] = None, **kwargs: Any) → Session

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

magpie.db.run_database_migration(settings: Optional[SettingsType] = None, db_session: Optional[Session] = None) → None

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

magpie.db.get_database_revision(db_session: Session) → Str

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

magpie.db.is_database_ready(db_session: Optional[Session] = None, container: Optional[AnySettingsContainer] = 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: SettingsType, db_session: Optional[Session] = None) → None

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

magpie.db.set_sqlalchemy_log_level(magpie_log_level: Union[Str, int]) → SettingsType

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

magpie.db.includeme(config)