Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
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()