Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
context_in_args: list of string. Input arguments from console.
Returns:
pypyr.context.Context() instance.
Raises:
AttributeError: parser specified on pipeline missing get_parsed_context
function.
"""
logger.debug("starting")
if 'context_parser' in pipeline:
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:
def assert_key_exists(self, key, caller):
"""Assert that context contains key.
Args:
key: validates that this key exists in context
caller: string. calling function or module name - this used to
construct error messages
Raises:
KeyNotInContextError: When key doesn't exist in context.
"""
assert key, ("key parameter must be specified.")
if key not in self:
raise KeyNotInContextError(
f"context['{key}'] doesn't exist. It must exist for {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}.")
key: dictionary key to retrieve.
Returns:
Formatted string.
Raises:
KeyNotInContextError: context[key] value contains {somekey} where
somekey does not exist in context dictionary.
"""
val = self[key]
if isinstance(val, str):
try:
return self.get_processed_string(val)
except KeyNotInContextError as err:
# Wrapping the KeyError into a less cryptic error for end-user
# friendliness
raise KeyNotInContextError(
f'Unable to format \'{val}\' at context[\'{key}\'], '
f'because {err}'
) from err
elif isinstance(val, SpecialTagDirective):
return val.get_value(self)
else:
# any sort of complex type will work with get_formatted_iterable.
return self.get_formatted_iterable(val)
Raises:
pypyr.errors.KeyNotInContextError: if ['pype']['name'] is missing.
pypyr.errors.KeyInContextHasNoValueError: if ['pype']['name'] exists but
is None.
"""
context.assert_key_has_value(key='pype', caller=__name__)
pype = context.get_formatted('pype')
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)
Returns:
Formatted string.
Raises:
KeyNotInContextError: context[key] has {somekey} where somekey does
not exist in context dictionary.
TypeError: Attempt operation on a non-string type.
"""
if isinstance(input_string, str):
try:
return self.get_processed_string(input_string)
except KeyNotInContextError as err:
# Wrapping the KeyError into a less cryptic error for end-user
# friendliness
raise KeyNotInContextError(
f'Unable to format \'{input_string}\' because {err}'
) from err
elif isinstance(input_string, SpecialTagDirective):
return input_string.get_value(self)
else:
raise TypeError(f"can only format on strings. {input_string} is a "
f"{type(input_string)} instead.")
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']
else:
has_default = False
default = None
return (env, key, has_default, default)
This step will run whatever combination of Get, Set and Unset you specify.
Regardless of combination, execution order is Get, Set, Unset.
"""
logger.debug("started")
assert context, f"context must have value for {__name__}"
context.assert_key_has_value('env', __name__)
found_get = env_get(context)
found_set = env_set(context)
found_unset = env_unset(context)
# at least 1 of envGet, envSet or envUnset must exist in context
if not (found_get or found_set or found_unset):
raise KeyNotInContextError(
"context must contain any combination of "
"env['get'], env['set'] or env['unset'] for "
f"{__name__}")
logger.debug("done")
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()