magpie.db ========= .. py:module:: magpie.db Attributes ---------- .. autoapisummary:: magpie.db.LOGGER Functions --------- .. autoapisummary:: magpie.db.get_constant magpie.db.get_logger magpie.db.get_settings magpie.db.get_settings_from_config_ini magpie.db.print_log magpie.db.raise_log magpie.db.get_db_url magpie.db.get_engine magpie.db.get_session_factory magpie.db.get_tm_session magpie.db.get_session_from_other magpie.db.get_db_session_from_settings magpie.db.get_db_session_from_config_ini magpie.db.get_connected_session magpie.db.run_database_migration magpie.db.get_database_revision magpie.db.is_database_ready magpie.db.run_database_migration_when_ready magpie.db.set_sqlalchemy_log_level magpie.db.includeme Module Contents --------------- .. py:function:: get_constant(constant_name: magpie.typedefs.Str, settings_container: Optional[magpie.typedefs.AnySettingsContainer] = None, settings_name: Optional[magpie.typedefs.Str] = None, default_value: Optional[magpie.typedefs.SettingValue] = 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 :paramref:`constant_name`: 1. search in :py:data:`MAGPIE_CONSTANTS` 2. search in settings if specified 3. search alternative setting names (see below) 4. search in :mod:`magpie.constants` definitions 5. search in environment variables Parameter :paramref:`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 :paramref:`settings_name` is provided as alternative name, it is used as is to search for results if :paramref:`constant_name` was not found. Otherwise, ``magpie.[variable_name]`` is used for additional search when the format ``MAGPIE_[VARIABLE_NAME]`` was used for :paramref:`constant_name` (i.e.: ``MAGPIE_ADMIN_USER`` will also search for ``magpie.admin_user`` and so on for corresponding constants). :param constant_name: key to search for a value :param settings_container: WSGI application settings container (if not provided, uses found one in current thread) :param settings_name: alternative name for `settings` if specified :param default_value: default value to be returned if not found anywhere, and exception raises are disabled. :param raise_not_set: raise an exception if the found key is ``None``, search until last case if others are ``None`` :param raise_missing: raise exception if key is not found anywhere :param print_missing: print message if key is not found anywhere, return ``None`` :param 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) :raises LookupError: if no appropriate value could be found from all search locations (according to options) .. py:function:: get_logger(name: magpie.typedefs.Str, level: Optional[int] = None, force_stdout: bool = None, message_format: Optional[magpie.typedefs.Str] = None, datetime_format: Optional[magpie.typedefs.Str] = 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``. .. py:function:: get_settings(container: Optional[magpie.typedefs.AnySettingsContainer], app: bool = False) -> magpie.typedefs.SettingsType Retrieve application settings from a supported container. :param container: supported container with a handle to application settings. :param app: allow retrieving from current thread registry if no container was defined. :return: found application settings dictionary. :raise TypeError: when no application settings could be found or unsupported container. .. py:function:: get_settings_from_config_ini(config_ini_path: magpie.typedefs.Str, ini_main_section_name: magpie.typedefs.Str = 'app:magpie_app') -> magpie.typedefs.SettingsType .. py:function:: print_log(msg: magpie.typedefs.Str, logger: Optional[logging.Logger] = 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``. .. py:function:: raise_log(msg: magpie.typedefs.Str, exception: Type[Exception] = Exception, logger: Optional[logging.Logger] = 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. .. py:data:: LOGGER .. py:function:: get_db_url(username: Optional[magpie.typedefs.Str] = None, password: Optional[magpie.typedefs.Str] = None, db_host: Optional[magpie.typedefs.Str] = None, db_port: Optional[Union[magpie.typedefs.Str, int]] = None, db_name: Optional[magpie.typedefs.Str] = None, settings: magpie.typedefs.AnySettingsContainer = None) -> magpie.typedefs.Str Retrieve the database connection URL with provided settings. .. py:function:: get_engine(container: Optional[magpie.typedefs.AnySettingsContainer] = None, prefix: magpie.typedefs.Str = 'sqlalchemy.', **kwargs: Any) -> sqlalchemy.engine.base.Engine .. py:function:: get_session_factory(engine) Create a new session with integrated thread-local scope for safe handling across application workers. .. py:function:: 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) .. py:function:: get_session_from_other(db_session) .. py:function:: get_db_session_from_settings(settings: Optional[magpie.typedefs.AnySettingsContainer] = None, **kwargs: Any) -> sqlalchemy.orm.session.Session .. py:function:: get_db_session_from_config_ini(config_ini_path, ini_main_section_name='app:magpie_app', settings_override=None) .. py:function:: 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. .. py:function:: run_database_migration(container: Optional[magpie.typedefs.AnySettingsContainer] = None, db_session: Optional[sqlalchemy.orm.session.Session] = None) -> None Runs database migration operations with :mod:`alembic`, using the provided session or a new engine connection. .. py:function:: get_database_revision(db_session: sqlalchemy.orm.session.Session) -> magpie.typedefs.Str Obtains the database revision number employed by :mod:`alembic` for schema migration. .. py:function:: is_database_ready(db_session: Optional[sqlalchemy.orm.session.Session] = None, container: Optional[magpie.typedefs.AnySettingsContainer] = None) -> bool Obtains the database status against expected table names to ensure it is ready for use. .. py:function:: run_database_migration_when_ready(settings: magpie.typedefs.SettingsType, db_session: Optional[sqlalchemy.orm.session.Session] = None) -> None Runs db migration if requested by config and need from revisions. .. py:function:: set_sqlalchemy_log_level(magpie_log_level: Union[magpie.typedefs.Str, int]) -> magpie.typedefs.SettingsType Suppresses :py:mod:`sqlalchemy` verbose logging if not in ``logging.DEBUG`` for Magpie. .. py:function:: includeme(config)