Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
parser_module_name = pipeline['context_parser']
logger.debug("context parser specified: %s", parser_module_name)
get_parsed_context = contextparser_cache.get_context_parser(
parser_module_name)
logger.debug("running parser %s", parser_module_name)
result_context = get_parsed_context(context_in_args)
logger.debug("context parse %s done", parser_module_name)
# Downstream steps likely to expect context not to be None, hence
# empty rather than None.
if result_context is None:
logger.debug(
"%s returned None. Using empty context instead",
parser_module_name
)
return pypyr.context.Context()
else:
return pypyr.context.Context(result_context)
else:
logger.debug("pipeline does not have custom context parser. Using "
"empty context.")
logger.debug("done")
# initialize to an empty dictionary because you want to be able to run
# with no context.
return pypyr.context.Context()
get_parsed_context = contextparser_cache.get_context_parser(
parser_module_name)
logger.debug("running parser %s", parser_module_name)
result_context = get_parsed_context(context_in_args)
logger.debug("context parse %s done", parser_module_name)
# Downstream steps likely to expect context not to be None, hence
# empty rather than None.
if result_context is None:
logger.debug(
"%s returned None. Using empty context instead",
parser_module_name
)
return pypyr.context.Context()
else:
return pypyr.context.Context(result_context)
else:
logger.debug("pipeline does not have custom context parser. Using "
"empty context.")
logger.debug("done")
# initialize to an empty dictionary because you want to be able to run
# with no context.
return pypyr.context.Context()
pipelinerunner.load_and_run_pipeline(
pipeline_name=pipeline_name,
pipeline_context_input=pipe_arg,
context=context,
parse_input=not skip_parse,
loader=loader,
groups=step_groups,
success_group=success_group,
failure_group=failure_group
)
else:
logger.info("pyping %s, without parent context.", pipeline_name)
if args:
child_context = Context(args)
else:
child_context = Context()
child_context.pipeline_name = pipeline_name
child_context.working_dir = context.working_dir
pipelinerunner.load_and_run_pipeline(
pipeline_name=pipeline_name,
pipeline_context_input=pipe_arg,
context=child_context,
parse_input=not skip_parse,
loader=loader,
groups=step_groups,
success_group=success_group,
failure_group=failure_group
)
loader (str): str. optional. Absolute name of pipeline loader module.
If not specified will use pypyr.pypeloaders.fileloader.
groups: list of str. step-group names to run in pipeline.
success_group: str. step-group name to run on success completion.
failure_group: str. step-group name to run on pipeline failure.
Returns:
None
"""
logger.debug("you asked to run pipeline: %s", pipeline_name)
logger.debug("you set the initial context to: %s", pipeline_context_input)
if context is None:
context = pypyr.context.Context()
context.pipeline_name = pipeline_name
context.working_dir = pypyr.moduleloader.get_working_directory()
# pipeline loading deliberately outside of try catch. The try catch will
# try to run a failure-handler from the pipeline, but if the pipeline
# doesn't exist there is no failure handler that can possibly run so this
# is very much a fatal stop error.
pipeline_definition = pipeline_cache.get_pipeline(
pipeline_name=pipeline_name,
loader=loader)
run_pipeline(
pipeline=pipeline_definition,
pipeline_context_input=pipeline_context_input,
context=context,
parse_input=parse_input,
# empty rather than None.
if result_context is None:
logger.debug(
"%s returned None. Using empty context instead",
parser_module_name
)
return pypyr.context.Context()
else:
return pypyr.context.Context(result_context)
else:
logger.debug("pipeline does not have custom context parser. Using "
"empty context.")
logger.debug("done")
# initialize to an empty dictionary because you want to be able to run
# with no context.
return pypyr.context.Context()