Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
_dump_workflow(executable, describe_output)
elif isinstance(executable, dxpy.DXGlobalWorkflow):
# Add inputs, outputs, stages. These fields contain region-specific values
# e.g. files or applets, that's why:
# * if the workflow is global, we will unpack the underlying workflow
# from the region of the current project context
# * if this is a regular, project-based workflow, we will just use
# its description (the describe_output that we already have)
# Underlying workflows are workflows stored in resource containers
# of the global workflow (one per each region the global workflow is
# enabled in). #TODO: add a link to documentation.
current_project = dxpy.WORKSPACE_ID
if not current_project:
raise DXError(
'A project needs to be selected to "dx get" a global workflow. You can use "dx select" to select a project')
region = dxpy.api.project_describe(current_project,
input_params={"fields": {"region": True}})["region"]
describe_output = executable.append_underlying_workflow_desc(describe_output, region)
_dump_workflow(executable, describe_output)
else:
_dump_app_or_applet(executable, omit_resources, describe_output)
except:
err_exit()
finally:
os.chdir(old_cwd)
"name": "Temporary build project for dx-build-app in {r}".format(r=region),
"region": region})["id"]
projects_by_region[region] = working_project
logger.debug("Created temporary project %s to build in" % (working_project,))
except:
# A /project/new request may fail if the requesting user is
# not authorized to create projects in a certain region.
delete_temporary_projects(projects_by_region.values())
err_exit()
else:
# Create a temp project
try:
working_project = dxpy.api.project_new({"name": "Temporary build project for dx-build-app"})["id"]
except:
err_exit()
region = dxpy.api.project_describe(working_project,
input_params={"fields": {"region": True}})["region"]
projects_by_region[region] = working_project
logger.debug("Created temporary project %s to build in" % (working_project,))
using_temp_project = True
elif mode == "app" and not dry_run:
# If we are not using temporary project(s) to build the executable,
# then we should have a project context somewhere.
try:
project = app_json.get("project", dxpy.WORKSPACE_ID)
region = dxpy.api.project_describe(project,
input_params={"fields": {"region": True}})["region"]
except Exception:
err_exit()
projects_by_region = {region: project}
# Perform check for existence of applet with same name in
# destination for case in which neither "-f" nor "-a" is
# given BEFORE uploading resources.
if mode == "applet" and not overwrite and not archive:
try:
dest_name = override_applet_name or app_json.get('name') or os.path.basename(os.path.abspath(src_dir))
except:
raise dxpy.app_builder.AppBuilderException("Could not determine applet name from specification + "
"(dxapp.json) or from working directory (%r)" % (src_dir,))
dest_folder = override_folder or app_json.get('folder') or '/'
if not dest_folder.endswith('/'):
dest_folder = dest_folder + '/'
dest_project = working_project if working_project else dxpy.WORKSPACE_ID
try:
region = dxpy.api.project_describe(dest_project,
input_params={"fields": {"region": True}})["region"]
except Exception:
err_exit()
projects_by_region = {region: dest_project}
for result in dxpy.find_data_objects(classname="applet", name=dest_name, folder=dest_folder,
project=dest_project, recurse=False):
dest_path = dest_folder + dest_name
msg = "An applet already exists at {} (id {}) and neither".format(dest_path, result["id"])
msg += " -f/--overwrite nor -a/--archive were given."
raise dxpy.app_builder.AppBuilderException(msg)
if dry_run:
# Set a dummy "projects_by_region" so that we can exercise the dry
# run flows for uploading resources bundles and applets below.
projects_by_region = {"dummy-cloud:dummy-region": "project-dummy"}
def _ensure_write_bufsize(self, **kwargs):
if self._write_bufsize is not None:
return
file_upload_params = dxpy.api.project_describe(
self.get_proj_id(),
{'fields': {'fileUploadParameters': True}},
**kwargs
)['fileUploadParameters']
self._empty_last_part_allowed = file_upload_params['emptyLastPartAllowed']
self._write_bufsize = _get_write_buf_size(self._write_buffer_size_hint,
file_upload_params,
self._expected_file_size,
self._file_is_mmapd)
try:
dnanexus_link = dxpy.describe(suggestion['$dnanexus_link'])
except dxpy.exceptions.DXAPIError as e:
if e.code == 404:
raise dxpy.app_builder.AppBuilderException(
'Suggested object {name} could not be found'.format(
name=suggestion['$dnanexus_link']))
except Exception as e:
raise dxpy.app_builder.AppBuilderException(str(e))
if 'value' in suggestion:
if '$dnanexus_link' in suggestion['value']:
# Check if we have JSON or string
if isinstance(suggestion['value']['$dnanexus_link'], dict):
if 'project' in suggestion['value']['$dnanexus_link']:
try:
dxpy.api.project_describe(suggestion['value']['$dnanexus_link']['project'])
except dxpy.exceptions.DXAPIError as e:
if e.code == 404:
logger.warn('Suggested project {name} does not exist, or not accessible by user'.format(
name=suggestion['value']['$dnanexus_link']['project']))
elif isinstance(suggestion['value']['$dnanexus_link'], basestring):
if suggestion['value']['$dnanexus_link'].startswith(('file-', 'record-', 'gtable-')):
try:
dnanexus_link = dxpy.describe(suggestion['value']['$dnanexus_link'])
except dxpy.exceptions.DXAPIError as e:
if e.code == 404:
raise dxpy.app_builder.AppBuilderException(
'Suggested object {name} could not be found'.format(
name=suggestion['value']['$dnanexus_link']))
except Exception as e:
raise dxpy.app_builder.AppBuilderException(str(e))
def _get_run_input(self, workflow_input, project=None, **kwargs):
"""
Checks the region in which the global workflow is run
and returns the input associated with the underlying workflow
from that region.
"""
region = dxpy.api.project_describe(project,
input_params={"fields": {"region": True}})["region"]
dxworkflow = self.get_underlying_workflow(region)
return dxworkflow._get_run_input(workflow_input, **kwargs)
def interactive_help(in_class, param_desc, prompt):
is_array = param_desc['class'].startswith("array:")
print_param_help(param_desc)
print()
array_help_str = ', or to finish the list of inputs'
if in_class in dx_data_classes:
# Class is some sort of data object
if dxpy.WORKSPACE_ID is not None:
proj_name = None
try:
proj_name = dxpy.api.project_describe(dxpy.WORKSPACE_ID)['name']
except:
pass
if proj_name is not None:
print('Your current working directory is ' + proj_name + ':' + dxpy.config.get('DX_CLI_WD', '/'))
while True:
print('Pick an option to find input data:')
try:
opt_num = pick(['List and choose from available data in the current project',
'List and choose from available data in the DNAnexus Reference Genomes project',
'Select another project to list and choose available data',
'Select an output from a previously-run job (current project only)',
'Return to original prompt (specify an ID or path directly)'])
except KeyboardInterrupt:
opt_num = 4
if opt_num == 0:
query_project = dxpy.WORKSPACE_ID
def _check_suggestions(app_json, publish=False):
"""
Examines the specified dxapp.json file and warns about any
violations of suggestions guidelines.
:raises: AppBuilderException for data objects that could not be found
"""
for input_field in app_json.get('inputSpec', []):
for suggestion in input_field.get('suggestions', []):
if 'project' in suggestion:
try:
project = dxpy.api.project_describe(suggestion['project'], {"permissions": True})
if 'PUBLIC' not in project['permissions'] and publish:
logger.warn('Project {name} NOT PUBLIC!'.format(name=project['name']))
except dxpy.exceptions.DXAPIError as e:
if e.code == 404:
logger.warn('Suggested project {name} does not exist, or not accessible by user'.format(
name=suggestion['project']))
if 'path' in suggestion:
try:
check_folder_exists(suggestion['project'], suggestion['path'], '')
except ResolutionError as e:
logger.warn('Folder {path} could not be found in project {project}'.format(
path=suggestion['path'], project=suggestion['project']))
if '$dnanexus_link' in suggestion:
if suggestion['$dnanexus_link'].startswith(('file-', 'record-', 'gtable-')):
try:
dnanexus_link = dxpy.describe(suggestion['$dnanexus_link'])