"""This modules describes the abstraction of a Galaxy profile.
This is a workspace with a specific default configuration and shed
tool setup. It is meant to be used with various serve commands.
"""
import json
import os
import shutil
import click
from gxjobconfinit import (
build_job_config,
ConfigArgs,
)
from planemo.database import (
database_source_class,
database_source_context,
)
from planemo.galaxy.api import test_credentials_valid
from .config import DATABASE_LOCATION_TEMPLATE
PROFILE_OPTIONS_JSON_NAME = "planemo_profile_options.json"
ALREADY_EXISTS_EXCEPTION = "Cannot create profile with name [%s], directory [%s] already exists."
def profile_exists(ctx, profile_name, **kwds):
"""Return a truthy value iff the specified profile already exists."""
profile_directory = _profile_directory(ctx, profile_name)
return os.path.exists(profile_directory)
[docs]
def list_profiles(ctx, **kwds):
"""Return a list of current profile names."""
return os.listdir(ctx.galaxy_profiles_directory)
[docs]
def delete_profile(ctx, profile_name, **kwds):
"""Delete profile with the specified name."""
profile_directory = _profile_directory(ctx, profile_name)
profile_options, _ = _load_profile_to_json(ctx, profile_name)
if profile_options["engine"] != "external_galaxy":
database_type = profile_options.get("database_type")
kwds["database_type"] = database_type
if database_type != "sqlite":
source_class = database_source_class(database_type)
for option in source_class.PROFILE_OPTIONS:
if option in profile_options:
kwds[option] = profile_options[option]
database_identifier = profile_options.get(
"database_identifier", _profile_to_database_identifier(profile_name)
)
with database_source_context(profile_directory=profile_directory, **kwds) as database_source:
database_source.delete_database(database_identifier)
shutil.rmtree(profile_directory)
[docs]
def create_profile(ctx, profile_name, **kwds):
"""Create a profile with the specified name."""
engine_type = kwds.get("engine", "galaxy")
profile_directory = _profile_directory(ctx, profile_name)
if profile_exists(ctx, profile_name, **kwds):
message = ALREADY_EXISTS_EXCEPTION % (profile_name, profile_directory)
raise click.ClickException(message)
os.makedirs(profile_directory)
try:
if engine_type == "docker_galaxy":
create_for_engine = _create_profile_docker
elif engine_type == "external_galaxy" or kwds.get("galaxy_url"):
create_for_engine = _create_profile_external
else:
create_for_engine = _create_profile_local
stored_profile_options = create_for_engine(ctx, profile_directory, profile_name, kwds)
profile_options_path = _stored_profile_options_path(profile_directory)
with open(profile_options_path, "w") as f:
json.dump(stored_profile_options, f)
except BaseException:
shutil.rmtree(profile_directory, ignore_errors=True)
raise
def _create_profile_docker(ctx, profile_directory, profile_name, kwds):
export_directory = os.path.join(profile_directory, "export")
os.makedirs(export_directory)
return {
"engine": "docker_galaxy",
}
def _create_profile_local(ctx, profile_directory, profile_name, kwds):
# A profile that named no backend gets its own sqlite file. Standing a postgres
# server up is only worth it when the user asked for one by name.
database_type = kwds.get("database_type") or "auto"
if database_type == "auto":
database_type = "sqlite"
if database_type == "sqlite":
database_location = os.path.join(profile_directory, "galaxy.sqlite")
database_connection = DATABASE_LOCATION_TEMPLATE % database_location
else:
database_identifier = _profile_to_database_identifier(profile_name)
with database_source_context(profile_directory=profile_directory, **kwds) as database_source:
database_source.create_database(database_identifier)
database_connection = database_source.sqlalchemy_url(database_identifier)
stored_options = {
"database_type": database_type,
"engine": "galaxy",
}
if database_source.store_connection_in_profile:
stored_options["database_connection"] = database_connection
else:
# This connection is derived again for each run so a managed server
# is started instead of being mistaken for an always-available one.
stored_options["database_identifier"] = database_identifier
stored_options.update(database_source.profile_options())
return stored_options
return {"database_type": database_type, "database_connection": database_connection, "engine": "galaxy"}
def _create_profile_external(ctx, profile_directory, profile_name, kwds):
url = kwds.get("galaxy_url")
api_key = kwds.get("galaxy_admin_key") or kwds.get("galaxy_user_key")
if test_credentials_valid(url=url, key=api_key, is_admin=kwds.get("galaxy_admin_key")):
return {
"galaxy_url": url,
"galaxy_user_key": kwds.get("galaxy_user_key"),
"galaxy_admin_key": kwds.get("galaxy_admin_key"),
"engine": "external_galaxy",
}
else:
raise ConnectionError("The credentials provided for an external Galaxy instance are not valid.")
[docs]
def ensure_profile(ctx, profile_name, **kwds):
"""Ensure a Galaxy profile exists and return profile defaults."""
if not profile_exists(ctx, profile_name, **kwds):
available_profiles = list_profiles(ctx, **kwds)
error_message = f"Profile '{profile_name}' does not exist."
if available_profiles:
error_message += f"\n\nAvailable profiles: {', '.join(available_profiles)}"
error_message += f"\n\nTo create a new profile, use: planemo profile_create {profile_name}"
else:
error_message += (
f"\n\nNo profiles found. To create a new profile, use: planemo profile_create {profile_name}"
)
raise click.UsageError(error_message)
return _profile_options(ctx, profile_name, **kwds)
def create_alias(ctx, alias, obj, profile_name, **kwds):
profile_options, profile_options_path = _load_profile_to_json(ctx, profile_name)
if profile_options.get("aliases"):
profile_options["aliases"][alias] = obj
else: # no aliases yet defined
profile_options["aliases"] = {alias: obj}
with open(profile_options_path, "w") as f:
json.dump(profile_options, f)
return 0
def list_alias(ctx, profile_name, **kwds):
profile_options, _ = _load_profile_to_json(ctx, profile_name)
return profile_options.get("aliases", {})
def delete_alias(ctx, alias, profile_name, **kwds):
profile_options, profile_options_path = _load_profile_to_json(ctx, profile_name)
if alias not in profile_options.get("aliases", {}):
return 1
else:
del profile_options["aliases"][alias]
with open(profile_options_path, "w") as f:
json.dump(profile_options, f)
return 0
def translate_alias(ctx, alias, profile_name):
if not profile_name:
return alias
aliases = _load_profile_to_json(ctx, profile_name)[0].get("aliases", {})
return aliases.get(alias, alias)
def _load_profile_to_json(ctx, profile_name):
if not profile_exists(ctx, profile_name):
raise click.ClickException("That profile does not exist. Create it with `planemo profile_create`")
profile_directory = _profile_directory(ctx, profile_name)
profile_options_path = _stored_profile_options_path(profile_directory)
with open(profile_options_path) as f:
profile_options = json.load(f)
return profile_options, profile_options_path
def _profile_options(ctx, profile_name, **kwds):
profile_directory = _profile_directory(ctx, profile_name)
profile_options, _ = _load_profile_to_json(ctx, profile_name)
if profile_options["engine"] == "docker_galaxy":
engine_options = dict(export_directory=os.path.join(profile_directory, "export"))
else:
file_path = os.path.join(profile_directory, "files")
shed_tool_path = os.path.join(profile_directory, "shed_tools")
shed_tool_conf = os.path.join(profile_directory, "shed_tool_conf.xml")
shed_tool_data_table_config = os.path.join(profile_directory, "shed_tool_data_table_conf.xml")
shed_data_manager_config = os.path.join(profile_directory, "shed_data_manager_conf.xml")
tool_dependency_dir = os.path.join(profile_directory, "deps")
mulled_resolution_cache_data_dir = os.path.join(profile_directory, "mulled_resolution_cache")
engine_options = dict(
file_path=file_path,
tool_dependency_dir=tool_dependency_dir,
shed_tool_conf=shed_tool_conf,
shed_tool_path=shed_tool_path,
shed_tool_data_table_config=shed_tool_data_table_config,
shed_data_manager_config=shed_data_manager_config,
galaxy_brand=profile_name,
mulled_resolution_cache_data_dir=mulled_resolution_cache_data_dir,
)
profile_options.update(engine_options)
profile_options["galaxy_brand"] = profile_name
return profile_options
def _profile_to_database_identifier(profile_name):
char_lst = [c if c.isalnum() else "_" for c in profile_name]
return "plnmoprof_%s" % "".join(char_lst)
def _stored_profile_options_path(profile_directory):
profile_options_path = os.path.join(profile_directory, PROFILE_OPTIONS_JSON_NAME)
return profile_options_path
def _profile_directory(ctx, profile_name):
return os.path.join(ctx.galaxy_profiles_directory, profile_name)
def initialize_job_config(ctx, profile_name, **kwds):
profile_directory = _profile_directory(ctx, profile_name)
job_config_path = os.path.join(profile_directory, "job_conf.yml")
if os.path.exists(job_config_path):
raise click.ClickException(f"File '{job_config_path}' already exists, exiting.")
init_config = ConfigArgs.from_dict(**kwds)
job_config = build_job_config(init_config)
with open(job_config_path, "w") as f:
f.write(job_config)
config, profile_config_path = _load_profile_to_json(ctx, profile_name)
config["job_config_file"] = os.path.abspath(job_config_path)
with open(profile_config_path, "w") as f:
json.dump(config, f)
return job_config_path
__all__ = (
"create_profile",
"delete_profile",
"ensure_profile",
"list_profiles",
)