Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Step that unsets env vars."""
import os
import pypyr.log.logger
# logger means the log level will be set correctly
logger = pypyr.log.logger.get_logger(__name__)
def run_step(context):
"""Unset $ENVs.
Context is a dictionary or dictionary-like. context is mandatory.
context['envUnset'] must exist. It's a list.
List items are the names of the $ENV values to unset.
For example, say input context is:
key1: value1
key2: value2
key3: value3
envUnset:
MYVAR1
"""Step that sets env vars from the pypyr context."""
import os
import pypyr.log.logger
# logger means the log level will be set correctly
logger = pypyr.log.logger.get_logger(__name__)
def run_step(context):
"""Set $ENVs from the pypyr context.
Context is a dictionary or dictionary-like. context is mandatory.
context['envSet'] must exist. It's a dictionary.
Values are the keys of the pypyr context values to write to $ENV.
Keys are the names of the $ENV values to which to write.
For example, say input context is:
key1: value1
key2: value2
key3: value3
envSet:
Args:
pipeline_name: string. Name of pipeline, sans .yaml at end.
pipeline_context_input: string. Initialize the pypyr context with this
string.
working_dir: path. looks for ./pipelines and modules in this directory.
log_level: int. Standard python log level enumerated value.
log_path: os.path. Append log to this path.
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
"""
pypyr.log.logger.set_root_logger(log_level, log_path)
logger.debug("starting pypyr")
# pipelines specify steps in python modules that load dynamically.
# make it easy for the operator so that the cwd is automatically included
# without needing to pip install a package 1st.
pypyr.moduleloader.set_working_directory(working_dir)
try:
load_and_run_pipeline(pipeline_name=pipeline_name,
pipeline_context_input=pipeline_context_input,
groups=groups,
success_group=success_group,
failure_group=failure_group)
except Stop:
logger.debug("Stop: stopped pypyr")