Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/planemo/galaxy/profiles.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:4f3585e2f14b
1 """This modules describes the abstraction of a Galaxy profile.
2
3 This is a workspace with a specific default configuration and shed
4 tool setup. It is meant to be used with various serve commands.
5 """
6 import json
7 import os
8 import shutil
9
10 from galaxy.util.commands import which
11
12 from planemo.database import create_database_source
13 from planemo.galaxy.api import test_credentials_valid
14 from .config import DATABASE_LOCATION_TEMPLATE
15
16 PROFILE_OPTIONS_JSON_NAME = "planemo_profile_options.json"
17 ALREADY_EXISTS_EXCEPTION = "Cannot create profile with name [%s], directory [%s] already exists."
18
19
20 def profile_exists(ctx, profile_name, **kwds):
21 """Return a truthy value iff the specified profile already exists."""
22 profile_directory = _profile_directory(ctx, profile_name)
23 return os.path.exists(profile_directory)
24
25
26 def list_profiles(ctx, **kwds):
27 """Return a list of current profile names."""
28 return os.listdir(ctx.galaxy_profiles_directory)
29
30
31 def delete_profile(ctx, profile_name, **kwds):
32 """Delete profile with the specified name."""
33 profile_directory = _profile_directory(ctx, profile_name)
34 profile_options = _read_profile_options(profile_directory)
35 profile_options, profile_options_path = _load_profile_to_json(ctx, profile_name)
36 if profile_options["engine"] != 'external_galaxy':
37 database_type = profile_options.get("database_type")
38 kwds["database_type"] = database_type
39 if database_type != "sqlite":
40 database_source = create_database_source(**kwds)
41 database_identifier = _profile_to_database_identifier(profile_name)
42 database_source.delete_database(
43 database_identifier,
44 )
45 shutil.rmtree(profile_directory)
46
47
48 def create_profile(ctx, profile_name, **kwds):
49 """Create a profile with the specified name."""
50 engine_type = kwds.get("engine", "galaxy")
51 profile_directory = _profile_directory(ctx, profile_name)
52 if profile_exists(ctx, profile_name, **kwds):
53 message = ALREADY_EXISTS_EXCEPTION % (
54 profile_name, profile_directory
55 )
56 raise Exception(message)
57
58 os.makedirs(profile_directory)
59
60 if engine_type == "docker_galaxy":
61 create_for_engine = _create_profile_docker
62 elif engine_type == "external_galaxy" or kwds.get("galaxy_url"):
63 create_for_engine = _create_profile_external
64 else:
65 create_for_engine = _create_profile_local
66
67 stored_profile_options = create_for_engine(ctx, profile_directory, profile_name, kwds)
68
69 profile_options_path = _stored_profile_options_path(profile_directory)
70 with open(profile_options_path, "w") as f:
71 json.dump(stored_profile_options, f)
72
73
74 def _create_profile_docker(ctx, profile_directory, profile_name, kwds):
75 export_directory = os.path.join(profile_directory, "export")
76 os.makedirs(export_directory)
77 return {
78 "engine": "docker_galaxy",
79 }
80
81
82 def _create_profile_local(ctx, profile_directory, profile_name, kwds):
83 database_type = kwds.get("database_type", "auto")
84 if database_type == "auto":
85 if which("psql"):
86 database_type = "postgres"
87 elif which("docker"):
88 database_type = "postgres_docker"
89 else:
90 database_type = "sqlite"
91
92 if database_type != "sqlite":
93 database_source = create_database_source(**kwds)
94 database_identifier = _profile_to_database_identifier(profile_name)
95 database_source.create_database(
96 database_identifier,
97 )
98 database_connection = database_source.sqlalchemy_url(database_identifier)
99 else:
100 database_location = os.path.join(profile_directory, "galaxy.sqlite")
101 database_connection = DATABASE_LOCATION_TEMPLATE % database_location
102
103 return {
104 "database_type": database_type,
105 "database_connection": database_connection,
106 "engine": "galaxy",
107 }
108
109
110 def _create_profile_external(ctx, profile_directory, profile_name, kwds):
111 url = kwds.get("galaxy_url")
112 api_key = kwds.get("galaxy_admin_key") or kwds.get("galaxy_user_key")
113 if test_credentials_valid(url=url, key=api_key, is_admin=kwds.get("galaxy_admin_key")):
114 return {
115 "galaxy_url": url,
116 "galaxy_user_key": kwds.get("galaxy_user_key"),
117 "galaxy_admin_key": kwds.get("galaxy_admin_key"),
118 "engine": "external_galaxy",
119 }
120 else:
121 raise ConnectionError('The credentials provided for an external Galaxy instance are not valid.')
122
123
124 def ensure_profile(ctx, profile_name, **kwds):
125 """Ensure a Galaxy profile exists and return profile defaults."""
126 if not profile_exists(ctx, profile_name, **kwds):
127 create_profile(ctx, profile_name, **kwds)
128
129 return _profile_options(ctx, profile_name, **kwds)
130
131
132 def create_alias(ctx, alias, obj, profile_name, **kwds):
133 profile_options, profile_options_path = _load_profile_to_json(ctx, profile_name)
134
135 if profile_options.get('aliases'):
136 profile_options['aliases'][alias] = obj
137 else: # no aliases yet defined
138 profile_options['aliases'] = {alias: obj}
139
140 with open(profile_options_path, 'w') as f:
141 json.dump(profile_options, f)
142
143 return 0
144
145
146 def list_alias(ctx, profile_name, **kwds):
147 profile_options, _ = _load_profile_to_json(ctx, profile_name)
148 return profile_options.get('aliases', {})
149
150
151 def delete_alias(ctx, alias, profile_name, **kwds):
152 profile_options, profile_options_path = _load_profile_to_json(ctx, profile_name)
153 if alias not in profile_options.get('aliases', {}):
154 return 1
155 else:
156 del profile_options['aliases'][alias]
157
158 with open(profile_options_path, 'w') as f:
159 json.dump(profile_options, f)
160
161 return 0
162
163
164 def translate_alias(ctx, alias, profile_name):
165 if not profile_name:
166 return alias
167 aliases = _load_profile_to_json(ctx, profile_name)[0].get('aliases', {})
168 return aliases.get(alias, alias)
169
170
171 def _load_profile_to_json(ctx, profile_name):
172 if not profile_exists(ctx, profile_name):
173 raise Exception("That profile does not exist. Create it with `planemo profile_create`")
174 profile_directory = _profile_directory(ctx, profile_name)
175 profile_options_path = _stored_profile_options_path(profile_directory)
176 with open(profile_options_path) as f:
177 profile_options = json.load(f)
178 return profile_options, profile_options_path
179
180
181 def _profile_options(ctx, profile_name, **kwds):
182 profile_directory = _profile_directory(ctx, profile_name)
183 profile_options = _read_profile_options(profile_directory)
184
185 if profile_options["engine"] == "docker_galaxy":
186 engine_options = dict(
187 export_directory=os.path.join(profile_directory, "export")
188 )
189 else:
190 file_path = os.path.join(profile_directory, "files")
191 shed_tool_path = os.path.join(profile_directory, "shed_tools")
192 shed_tool_conf = os.path.join(profile_directory, "shed_tool_conf.xml")
193 tool_dependency_dir = os.path.join(profile_directory, "deps")
194
195 engine_options = dict(
196 file_path=file_path,
197 tool_dependency_dir=tool_dependency_dir,
198 shed_tool_conf=shed_tool_conf,
199 shed_tool_path=shed_tool_path,
200 galaxy_brand=profile_name,
201 )
202 profile_options.update(engine_options)
203 profile_options["galaxy_brand"] = profile_name
204 return profile_options
205
206
207 def _profile_to_database_identifier(profile_name):
208 char_lst = [c if c.isalnum() else "_" for c in profile_name]
209 return "plnmoprof_%s" % "".join(char_lst)
210
211
212 def _read_profile_options(profile_directory):
213 profile_options_path = _stored_profile_options_path(profile_directory)
214 with open(profile_options_path, "r") as f:
215 profile_options = json.load(f)
216 return profile_options
217
218
219 def _stored_profile_options_path(profile_directory):
220 profile_options_path = os.path.join(
221 profile_directory, PROFILE_OPTIONS_JSON_NAME
222 )
223 return profile_options_path
224
225
226 def _profile_directory(ctx, profile_name):
227 return os.path.join(ctx.galaxy_profiles_directory, profile_name)
228
229
230 __all__ = (
231 "create_profile",
232 "delete_profile",
233 "ensure_profile",
234 "list_profiles",
235 )