Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
available workflows.
"""
workflow_name = workflow_object.__class__.__name__
workflow = select_workflow_by_name(session, workflow_name)
if workflow is None:
session.add(Workflow(name=workflow_name))
workflow = select_workflow_by_name(session, workflow_name)
for element_name in workflow_object.element_names:
tokens = element_name.split('.')
element_filename = tokens[0]
# inferring that file is code if there is no extension
if len(tokens) > 2:
raise ValueError('File name {} should contain at most one "."'
.format(element_name))
element_file_extension_name = tokens[1] if len(tokens) == 2 else 'py'
extension = select_extension_by_name(session,
element_file_extension_name)
if extension is None:
raise ValueError('Unknown extension {}.'
.format(element_file_extension_name))
type_extension = select_submission_type_extension_by_extension(
session, extension
)
if type_extension is None:
raise ValueError('Unknown file type {}.'
.format(element_file_extension_name))
workflow_element_type = select_workflow_element_type_by_name(
session, element_filename
)
if workflow_element_type is None:
workflow_element_type = WorkflowElementType(
def add_extension(session, name):
"""Adding a new extension, e.g., 'py'.
Parameters
----------
session : :class:`sqlalchemy.orm.Session`
The session to directly perform the operation on the database.
name : str
The name of the extension to add if it does not exist.
"""
extension = select_extension_by_name(session, name)
if extension is None:
extension = Extension(name=name)
logger.info('Adding {}'.format(extension))
session.add(extension)
session.commit()
type_name : str
The file type.
extension_name : str
The extension name.
Notes
-----
Should be preceded by adding submission file types and extensions.
"""
type_extension = select_submission_type_extension_by_name(
session, type_name, extension_name
)
if type_extension is None:
submission_file_type = select_submission_file_type_by_name(session,
type_name)
extension = select_extension_by_name(session, extension_name)
type_extension = SubmissionFileTypeExtension(
type=submission_file_type,
extension=extension
)
logger.info('Adding {}'.format(type_extension))
session.add(type_extension)
session.commit()
for filename, extension in files_type_extension
if filename == workflow_element.name]
)
except ValueError as e:
session.rollback()
if 'values to unpack' in str(e):
# no file matching the workflow element
raise MissingSubmissionFileError(
'No file corresponding to the workflow element "{}"'
.format(workflow_element)
)
raise
# check that files have the correct extension ...
for extension_name in deposited_extensions:
extension = select_extension_by_name(session, extension_name)
if extension is not None:
break
# ... otherwise we raise an error
else:
session.rollback()
raise MissingExtensionError(
'All extensions "{}" are unknown for the submission "{}".'
.format(", ".join(deposited_extensions), submission_name)
)
# check if it is a resubmission
submission_file = (session.query(SubmissionFile)
.filter(SubmissionFile.workflow_element ==
workflow_element)
.filter(SubmissionFile.submission ==
submission)
Parameters
----------
session : :class:`sqlalchemy.orm.Session`
The session to directly perform the operation on the database.
extension_name : str or None
The name of the extension to query. If None, all the extensions will be
queried.
Returns
-------
extension : :class:`ramp_database.model.Extension` or list of \
:class:`ramp_database.model.Extension`
The queried extension.
"""
return select_extension_by_name(session, extension_name)