Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
args = pype.get('args', None)
if args is not None and not isinstance(args, dict):
raise ContextError(
"pypyr.steps.pype 'args' in the 'pype' context item "
"must be a dict.")
if args and 'useParentContext' not in pype:
use_parent_context = False
else:
use_parent_context = pype.get('useParentContext', True)
out = pype.get('out', None)
if out and use_parent_context:
raise ContextError(
"pypyr.steps.pype pype.out is only relevant if useParentContext "
"= False. If you're using the parent context, no need to have out "
"args since their values will already be in context. If you're "
"NOT using parent context and you've specified pype.args, just "
"leave off the useParentContext key and it'll default to False "
"under the hood, or set it to False yourself if you keep it in.")
pipe_arg = pype.get('pipeArg', None)
skip_parse = pype.get('skipParse', True)
raise_error = pype.get('raiseError', True)
loader = pype.get('loader', None)
groups = pype.get('groups', None)
if isinstance(groups, str):
groups = [groups]
success_group = pype.get('success', None)
self.cmd_text = context.get_formatted_string(run_string)
is_save = cmd_config.get('save', False)
self.is_save = context.get_formatted_as_type(is_save,
out_type=bool)
cwd_string = cmd_config.get('cwd', None)
if cwd_string:
self.cwd = context.get_formatted_string(cwd_string)
self.logger.debug("Processing command string in dir "
"%s: %s", self.cwd, run_string)
else:
self.cwd = None
self.logger.debug("Processing command string: %s", run_string)
else:
raise ContextError(f"{name} cmd config should be either a simple "
"string cmd='mycommandhere' or a dictionary "
try:
pipeline_name = pype['name']
if pipeline_name is None:
raise KeyInContextHasNoValueError(
"pypyr.steps.pype ['pype']['name'] exists but is empty.")
except KeyError as err:
raise KeyNotInContextError(
"pypyr.steps.pype missing 'name' in the 'pype' context item. "
"You need to specify the pipeline name to run another "
"pipeline.") from err
args = pype.get('args', None)
if args is not None and not isinstance(args, dict):
raise ContextError(
"pypyr.steps.pype 'args' in the 'pype' context item "
"must be a dict.")
if args and 'useParentContext' not in pype:
use_parent_context = False
else:
use_parent_context = pype.get('useParentContext', True)
out = pype.get('out', None)
if out and use_parent_context:
raise ContextError(
"pypyr.steps.pype pype.out is only relevant if useParentContext "
"= False. If you're using the parent context, no need to have out "
"args since their values will already be in context. If you're "
"NOT using parent context and you've specified pype.args, just "
"leave off the useParentContext key and it'll default to False "
Raises:
KeyNotInContextError: Key doesn't exist
KeyInContextHasNoValueError: context[key] is None
AssertionError: if key is None
"""
assert parent, ("parent parameter must be specified.")
assert child, ("child parameter must be specified.")
self.assert_key_has_value(parent, caller)
try:
child_exists = child in self[parent]
except TypeError as err:
# This happens if parent isn't iterable
raise ContextError(
f"context['{parent}'] must be iterable and contain '{child}' "
f"for {caller}. {err}") from err
if child_exists:
if self[parent][child] is None:
raise KeyInContextHasNoValueError(
f"context['{parent}']['{child}'] must have a value for "
f"{caller}.")
else:
raise KeyNotInContextError(
f"context['{parent}']['{child}'] doesn't "
f"exist. It must exist for {caller}.")
Args:
out. str or dict or list. Pass a string for a single
key to grab from child context, a list of string for a list
of keys to grab from child context, or a dict where you map
'parent-key-name': 'child-key-name'.
parent_context: parent Context. destination context.
child_context: write from this context to the parent.
"""
if isinstance(out, str):
save_me = {out: out}
elif isinstance(out, list):
save_me = {k: k for k in out}
elif isinstance(out, dict):
save_me = out
else:
raise ContextError("pypyr.steps.pype pype.out should be a string, or "
f"a list or a dict. Instead, it's a {type(out)}")
for parent_key, child_key in save_me.items():
logger.debug(
"setting parent context %s to value from child context %s",
parent_key,
child_key)
parent_context[parent_key] = child_context.get_formatted(child_key)
if not assert_result:
if is_equals_there:
# emit type to help user, but not the actual field contents.
type_this = (
type(context.get_formatted_iterable(assert_this)).__name__)
type_equals = (
type(context.get_formatted_iterable(assert_equals)).__name__)
error_text = (
f"assert assert['this'] is of type {type_this} "
f"and does not equal assert['equals'] of type {type_equals}.")
else:
# if it's a bool it's presumably not a sensitive value.
error_text = (
f"assert {assert_this} evaluated to False.")
raise ContextError(error_text)
logger.debug("done")
get_item: dict. contains keys env/key/default
Returns:
(env, key, has_default, default) tuple, where
env: str. env var name.
key: str. save env value to this context key.
has_default: bool. True if default specified.
default: the value of default, if specified.
Raises:
ContextError: envGet is not a list of dicts.
KeyNotInContextError: If env or key not found in get_config.
"""
if not isinstance(get_item, dict):
raise ContextError('envGet must contain a list of dicts.')
env = get_item.get('env', None)
if not env:
raise KeyNotInContextError(
'context envGet[env] must exist in context for envGet.')
key = get_item.get('key', None)
if not key:
raise KeyNotInContextError(
'context envGet[key] must exist in context for envGet.')
if 'default' in get_item:
has_default = True
default = get_item['default']
"""Base class for all pypyr exceptions."""
class ContextError(Error):
"""Error in the pypyr context."""
class HandledError(Error):
"""Error that has already been saved to errors context collection"""
class KeyInContextHasNoValueError(ContextError):
"""pypyr context[key] doesn't have a value."""
class KeyNotInContextError(ContextError, KeyError):
"""Key not found in the pypyr context."""
def __str__(self):
"""KeyError has custom error formatting, avoid this behaviour."""
return super(Exception, self).__str__()
class LoopMaxExhaustedError(Error):
"""Max attempts reached during looping."""
class PipelineDefinitionError(Error):
"""Pipeline definition incorrect. Likely a yaml error."""
class PipelineNotFoundError(Error):