Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def delete_problem(session, problem_name):
"""Delete a problem from the database.
Parameters
----------
session : :class:`sqlalchemy.orm.Session`
The session to directly perform the operation on the database.
problem_name : str
The name of the problem to remove.
"""
problem = select_problem_by_name(session, problem_name)
if problem is None:
raise NoResultFound('No result found for "{}" in Problem table'
.format(problem_name))
for event in problem.events:
delete_event(session, event.name)
session.delete(problem)
session.commit()
"""Add relationship between a keyword and a problem.
Parameters
----------
session : :class:`sqlalchemy.orm.Session`
The session to directly perform the operation on the database.
problem_name : str
The name of the problem.
keyword_name : str
The name of the keyword.
description : None or str, default is None
A particular description of the keyword of the particular problem.
force : bool, default is False
Whether or not to overwrite the relationship.
"""
problem = select_problem_by_name(session, problem_name)
keyword = get_keyword_by_name(session, keyword_name)
problem_keyword = (session.query(ProblemKeyword)
.filter_by(problem=problem, keyword=keyword)
.one_or_none())
if problem_keyword is not None:
if not force:
raise ValueError(
'Attempting to update an existing problem-keyword '
'relationship. Use "force=True" if you want to overwrite the '
'relationship.'
)
problem_keyword.description = description
else:
problem_keyword = ProblemKeyword(
problem=problem, keyword=keyword, description=description
)
Parameters
----------
session : :class:`sqlalchemy.orm.Session`
The session to directly perform the operation on the database.
problem_name : str or None
The name of the problem to query. If None, all the problems will be
queried.
Returns
-------
problem : :class:`ramp_database.model.Problem` or list of \
:class:`ramp_database.model.Problem`
The queried problem.
"""
return select_problem_by_name(session, problem_name)
Parameters
----------
session : :class:`sqlalchemy.orm.Session`
The session to directly perform the operation on the database.
problem_name : str
The name of the problem.
keyword_name : str
The name of the keyword.
Returns
-------
problem_keyword : :class:`ramp_database.model.ProblemKeyword`
The problem-keyword relationship.
"""
problem = select_problem_by_name(session, problem_name)
keyword = session.query(Keyword).filter_by(name=keyword_name).one_or_none()
return (session.query(ProblemKeyword)
.filter_by(problem=problem, keyword=keyword)
.one_or_none())
The session to directly perform the operation on the database.
problem_name : str
The name of the problem to register in the database.
kit_dir : str
The directory where the RAMP kit are located. It will corresponds to
the key `ramp_kit_dir` of the dictionary created with
:func:`ramp_utils.generate_ramp_config`.
data_dir : str
The directory where the RAMP data are located. It will corresponds to
the key `ramp_data_dir` of the dictionary created with
:func:`ramp_utils.generate_ramp_config`.
force : bool, default is False
Whether to force add the problem. If ``force=False``, an error is
raised if the problem was already in the database.
"""
problem = select_problem_by_name(session, problem_name)
problem_kit_path = kit_dir
if problem is not None:
if not force:
raise ValueError('Attempting to overwrite a problem and '
'delete all linked events. Use"force=True" '
'if you want to overwrite the problem and '
'delete the events.')
delete_problem(session, problem_name)
# load the module to get the type of workflow used for the problem
problem_module = import_module_from_source(
os.path.join(problem_kit_path, 'problem.py'), 'problem')
add_workflow(session, problem_module.workflow)
problem = Problem(name=problem_name, path_ramp_kit=kit_dir,
path_ramp_data=data_dir, session=session)
logger.info('Adding {}'.format(problem))