Source code for galaxy.model.mapping

import logging
from threading import local
from typing import (
    Optional,
    Type,
)

from galaxy import model
from galaxy.config import GalaxyAppConfiguration
from galaxy.model import mapper_registry
from galaxy.model.base import SharedModelMapping
from galaxy.model.orm.engine_factory import build_engine
from galaxy.model.security import GalaxyRBACAgent
from galaxy.model.triggers.update_audit_table import install as install_timestamp_triggers

log = logging.getLogger(__name__)

metadata = mapper_registry.metadata


[docs]class GalaxyModelMapping(SharedModelMapping):
    security_agent: GalaxyRBACAgent
    thread_local_log: Optional[local]
    User: Type
    GalaxySession: Type


[docs]def init(
    file_path,
    url,
    engine_options=None,
    create_tables=False,
    map_install_models=False,
    database_query_profiling_proxy=False,
    object_store=None,
    trace_logger=None,
    use_pbkdf2=True,
    slow_query_log_threshold=0,
    thread_local_log: Optional[local] = None,
    log_query_counts=False,
) -> GalaxyModelMapping:
    # Build engine
    engine = build_engine(
        url,
        engine_options,
        database_query_profiling_proxy,
        trace_logger,
        slow_query_log_threshold,
        thread_local_log=thread_local_log,
        log_query_counts=log_query_counts,
    )

    # Create tables if needed
    if create_tables:
        mapper_registry.metadata.create_all(bind=engine)
        create_additional_database_objects(engine)
        if map_install_models:
            from galaxy.model.tool_shed_install import mapping as install_mapping  # noqa: F401

            install_mapping.create_database_objects(engine)

    # Configure model, build ModelMapping
    return configure_model_mapping(file_path, object_store, use_pbkdf2, engine, map_install_models, thread_local_log)


[docs]def create_additional_database_objects(engine):
    install_timestamp_triggers(engine)


[docs]def configure_model_mapping(
    file_path: str,
    object_store,
    use_pbkdf2,
    engine,
    map_install_models,
    thread_local_log,
) -> GalaxyModelMapping:
    _configure_model(file_path, object_store, use_pbkdf2)
    return _build_model_mapping(engine, map_install_models, thread_local_log)


def _configure_model(file_path: str, object_store, use_pbkdf2) -> None:
    model.Dataset.file_path = file_path
    model.Dataset.object_store = object_store
    model.User.use_pbkdf2 = use_pbkdf2


def _build_model_mapping(engine, map_install_models, thread_local_log) -> GalaxyModelMapping:
    model_modules = [model]
    if map_install_models:
        from galaxy.model import tool_shed_install

        model_modules.append(tool_shed_install)

    model_mapping = GalaxyModelMapping(model_modules, engine)
    model_mapping.security_agent = GalaxyRBACAgent(model_mapping)
    model_mapping.thread_local_log = thread_local_log
    return model_mapping


[docs]def init_models_from_config(
    config: GalaxyAppConfiguration, map_install_models=False, object_store=None, trace_logger=None
):
    model = init(
        config.file_path,
        config.database_connection,
        config.database_engine_options,
        map_install_models=map_install_models,
        database_query_profiling_proxy=config.database_query_profiling_proxy,
        object_store=object_store,
        trace_logger=trace_logger,
        use_pbkdf2=config.get_bool("use_pbkdf2", True),
        slow_query_log_threshold=config.slow_query_log_threshold,
        thread_local_log=config.thread_local_log,
        log_query_counts=config.database_log_query_counts,
    )
    return model