magpie.db

Attributes

LOGGER

Functions

get_constant(→ magpie.typedefs.SettingValue)	Search in order for matched value of constant_name:
get_logger(→ logging.Logger)	Immediately sets the logger level to avoid duplicate log outputs from the root logger and this logger when
get_settings(→ magpie.typedefs.SettingsType)	Retrieve application settings from a supported container.
get_settings_from_config_ini(...)
print_log(→ None)	Logs the requested message to the logger and optionally enforce printing to the console according to configuration
raise_log(→ NoReturn)	Logs the provided message to the logger and raises the corresponding exception afterwards.
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.get_constant(constant_name: magpie.typedefs.Str, settings_container: magpie.typedefs.AnySettingsContainer | None = None, settings_name: magpie.typedefs.Str | None = None, default_value: magpie.typedefs.SettingValue | None = None, raise_not_set: bool = True, raise_missing: bool = True, print_missing: bool = False, empty_missing: bool = False) → magpie.typedefs.SettingValue

Search in order for matched value of constant_name:

1. search in MAGPIE_CONSTANTS

2. search in settings if specified

3. search alternative setting names (see below)

4. search in magpie.constants definitions

5. search in environment variables

Parameter constant_name is expected to have the format MAGPIE_[VARIABLE_NAME] although any value can be passed to retrieve generic settings from all above-mentioned search locations.

If settings_name is provided as alternative name, it is used as is to search for results if constant_name was not found. Otherwise, magpie.[variable_name] is used for additional search when the format MAGPIE_[VARIABLE_NAME] was used for constant_name (i.e.: MAGPIE_ADMIN_USER will also search for magpie.admin_user and so on for corresponding constants).

Parameters:

• constant_name – key to search for a value

• settings_container – WSGI application settings container (if not provided, uses found one in current thread)

• settings_name – alternative name for settings if specified

• default_value – default value to be returned if not found anywhere, and exception raises are disabled.

• raise_not_set – raise an exception if the found key is None, search until last case if others are None

• raise_missing – raise exception if key is not found anywhere

• print_missing – print message if key is not found anywhere, return None

• empty_missing – consider an empty value for an existing key as if it was missing (i.e.: as if not set).

Returns:

found value or default_value

Raises:

• ValueError – if resulting value is invalid based on options (by default raise missing/empty/None value)

• LookupError – if no appropriate value could be found from all search locations (according to options)

magpie.db.get_logger(name: magpie.typedefs.Str, level: int | None = None, force_stdout: bool = None, message_format: magpie.typedefs.Str | None = None, datetime_format: magpie.typedefs.Str | None = None) → logging.Logger

Immediately sets the logger level to avoid duplicate log outputs from the root logger and this logger when level is logging.NOTSET.

magpie.db.get_settings(container: magpie.typedefs.AnySettingsContainer | None, app: bool = False) → magpie.typedefs.SettingsType

Retrieve application settings from a supported container.

Parameters:

• container – supported container with a handle to application settings.

• app – allow retrieving from current thread registry if no container was defined.

Returns:

found application settings dictionary.

Raises:

TypeError – when no application settings could be found or unsupported container.

magpie.db.get_settings_from_config_ini(config_ini_path: magpie.typedefs.Str, ini_main_section_name: magpie.typedefs.Str = 'app:magpie_app') → magpie.typedefs.SettingsType

magpie.db.print_log(msg: magpie.typedefs.Str, logger: logging.Logger | None = None, level: int = logging.INFO, **kwargs: Any) → None

Logs the requested message to the logger and optionally enforce printing to the console according to configuration value defined by MAGPIE_LOG_PRINT.

magpie.db.raise_log(msg: magpie.typedefs.Str, exception: Type[Exception] = Exception, logger: logging.Logger | None = None, level: int = logging.ERROR) → NoReturn

Logs the provided message to the logger and raises the corresponding exception afterwards.

Raises:

exception – whichever exception provided is raised systematically after logging.

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)