Source code for galaxy.model.orm.scripts

"""
Code to support database helper scripts (create_toolshed_db.py, migrate_toolshed_db.py, etc...).
"""

import argparse
import logging
import os
import sys

import alembic.config

from galaxy.model.migrations import (
    GXY,
    TSI,
)
from galaxy.model.migrations.scripts import get_configuration
from galaxy.util.path import get_ext
from galaxy.util.properties import (
    find_config_file,
    get_data_dir,
    load_app_properties,
)
from galaxy.util.script import populate_config_args

log = logging.getLogger(__name__)

DEFAULT_CONFIG_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, "config", "sample"))
DEFAULT_CONFIG_NAMES = ["galaxy", "universe_wsgi"]
DEFAULT_CONFIG_PREFIX = ""
DEFAULT_DATABASE = "galaxy"

DATABASE = {
    "galaxy": {
        "default_sqlite_file": "universe.sqlite",
        "config_override": "GALAXY_CONFIG_",
    },
    "tool_shed": {
        "repo": "tool_shed/webapp/model/migrate",
        "config_names": ["tool_shed", "tool_shed_wsgi"],
        "default_sqlite_file": "community.sqlite",
        "config_override": "TOOL_SHED_CONFIG_",
        "config_section": "tool_shed",
    },
    "install": {
        "config_prefix": "install_",
        "default_sqlite_file": "install.sqlite",
        "config_override": "GALAXY_INSTALL_CONFIG_",
    },
}


def _read_model_arguments(argv, use_argparse=False):
    if use_argparse:
        parser = argparse.ArgumentParser()
        parser.add_argument(
            "database",
            metavar="DATABASE",
            type=str,
            default="galaxy",
            nargs="?",
            help="database to target (galaxy, tool_shed, install)",
        )
        populate_config_args(parser)
        args = parser.parse_args(argv[1:] if argv else [])
        return args.config_file, args.config_section, args.database
    else:
        config_file = None
        for arg in ["-c", "--config", "--config-file"]:
            if arg in argv:
                pos = argv.index(arg)
                argv.pop(pos)
                config_file = argv.pop(pos)
        config_section = None
        if "--config-section" in argv:
            pos = argv.index("--config-section")
            argv.pop(pos)
            config_section = argv.pop(pos)
        if argv and (argv[-1] in DATABASE):
            database = argv.pop()  # database name tool_shed, galaxy, or install.
        else:
            database = "galaxy"
        return config_file, config_section, database


[docs]def get_config(argv, use_argparse=True, cwd=None): """ Read sys.argv and parse out repository of migrations and database url. >>> import os >>> from configparser import ConfigParser >>> from shutil import rmtree >>> from tempfile import mkdtemp >>> config_dir = mkdtemp() >>> os.makedirs(os.path.join(config_dir, 'config')) >>> def write_ini(path, property, value): ... p = ConfigParser() ... p.add_section('app:main') ... p.set('app:main', property, value) ... with open(os.path.join(config_dir, 'config', path), 'w') as f: p.write(f) >>> write_ini('tool_shed.ini', 'database_connection', 'sqlite:///pg/testdb1') >>> config = get_config(['manage_db.py', 'tool_shed'], cwd=config_dir) >>> config['repo'].endswith('tool_shed/webapp/model/migrate') True >>> config['db_url'] 'sqlite:///pg/testdb1' >>> write_ini('galaxy.ini', 'data_dir', '/moo') >>> config = get_config(['manage_db.py'], cwd=config_dir) >>> uri_with_env = os.getenv("GALAXY_TEST_DBURI", "sqlite:////moo/universe.sqlite?isolation_level=IMMEDIATE") >>> config['db_url'] == uri_with_env True >>> rmtree(config_dir) """ config_file, config_section, database = _read_model_arguments(argv, use_argparse=use_argparse) database_defaults = DATABASE[database] if config_file is None: config_names = database_defaults.get("config_names", DEFAULT_CONFIG_NAMES) if cwd: cwd = [cwd, os.path.join(cwd, "config")] else: cwd = [DEFAULT_CONFIG_DIR] config_file = find_config_file(config_names, dirs=cwd) repo = database_defaults.get("repo") if repo: repo = os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir, repo) config_prefix = database_defaults.get("config_prefix", DEFAULT_CONFIG_PREFIX) config_override = database_defaults.get("config_override", "GALAXY_CONFIG_") default_sqlite_file = database_defaults["default_sqlite_file"] if config_section is None: if not config_file or get_ext(config_file, ignore="sample") == "yaml": config_section = database_defaults.get("config_section", None) else: # Just use the default found by load_app_properties. config_section = None properties = load_app_properties( config_file=config_file, config_prefix=config_override, config_section=config_section ) if (f"{config_prefix}database_connection") in properties: db_url = properties[f"{config_prefix}database_connection"] else: db_url = f"sqlite:///{os.path.join(get_data_dir(properties), default_sqlite_file)}?isolation_level=IMMEDIATE" install_database_connection = properties.get("install_database_connection") return dict( db_url=db_url, repo=repo, config_file=config_file, database=database, install_database_connection=install_database_connection, )
[docs]def manage_db(): # This is a duplicate implementation of scripts/migrate_db.py. # See run_alembic.sh for usage. def _insert_x_argument(key, value): sys.argv.insert(1, f"{key}={value}") sys.argv.insert(1, "-x") gxy_config, tsi_config, _ = get_configuration(sys.argv, os.getcwd()) _insert_x_argument("tsi_url", tsi_config.url) _insert_x_argument("gxy_url", gxy_config.url) if "heads" in sys.argv and "upgrade" in sys.argv: i = sys.argv.index("heads") sys.argv[i] = f"{GXY}@head" alembic.config.main() sys.argv[i] = f"{TSI}@head" alembic.config.main() else: alembic.config.main()