silo.low_code_steps.standard package

class silo.low_code_steps.standard.CSVParser

Bases: object

get_opts(action_arg)

class silo.low_code_steps.standard.JMESPathParser

Bases: object

bad_action_arg_msg = "Action_argument 'index=true' was set; however, the result of the query did not yield _index and _value keys in its result. Please provide the step with a similar formatted value path: \n jmespath:\n index: true\n value: path.to[].{ _index: index.path , _value: value.path }\n "

get_paths(action_arg, context_logger)

Turns the given action argument into a JMESPathString type

Transform a action argument (arg derived from snippet arg) and turns it into the index and a value jmespath string. Provides argument checking.

Parameters

action_arg (dict) – step parameters provided by the snippet argument
context_logger (logger) – Logger for the step

Returns

index boolean, value str

Return type

JMESPathString()

Raises

ValueError – Raised if an invalid argument was given

static handle_merge_for_list_of_dicts(value_matches)

Handles merging JMESPath matches contained in an iterable

Parameters: value_matches (dict) – JMESPath matches
Returns: merged index value pairs
Return type: dict

static merge_index_matches_to_dict(value_matches)

Merges index and value JMESPath Expressions into key-value pairs

When a multiselect hash expression is used, a list of matches will be used to form key/value pairs which are then able to be turned into a dictionary. The simplest case is when the full path of the index and value match: path.to[n].{ _index: index_item, _value: value_item }

In this case, we know that the index and value are 1:1 and we can create the following dict entry:

d[index] = value

The more complicated cases are some varient of a sub-entry: path.to[n].sub.value and path.to[n].index Here, we see that path.to[n].sub != path.to[n] Thus, we remove additional levels on the value path to see if we can find a match. In this case, we do:

path.to[n].{ _index: index, _value: sub.value }

Parameters: value_matches (any) – result of a JMESPath multiselect hash search
Returns: Key-Value pairs of index and value expression results
Return type: dict or None

class silo.low_code_steps.standard.JSONPathParser

Bases: object

format_parsed_values(match)

Takes matched values and extracts the values from it. If there is only one value in the: matched results, return just that object, not a list.

Parameters: match (list) – List of DataumInContext values
Returns: The value from the matched items
Return type: list

get_paths(action_arg, context_logger)

Turns action argument into index and value jsonpath strings.

Takes the action argument (arg derived from snippet arg) and turns it into: the index and value jsonpath strings. This is a helper to break down the function

Parameters

action_arg (str) – The jsonpath string provided by the snippet
context_logger (logger) – Logger for the step

Returns

index JSONPath, value JSONPath

Return type

JpathString()

Raises

ValueError – Raised if an invalid argument was given

merge_index_matches_to_dict(value_matches, index_matches)

If two jpaths were used, you get two lists of matches that can be used as key/value

pairs which are then able to be turned in to our tuples. The simplest case is when the full path of the index and value basically match: path.to[n].value and path.to[n].index

In this case, we knowt hat they are the same: path.to[n] == path.to[n] in this case, we can take the value and index and create the dict entry: d[index] = value

The more complicated cases are some varient of a sub-entry: path.to[n].sub.value and path.to[n].index Here, we see that path.to[n].sub != path.to[n] Thus, we remove additional levels on the value path to see if we can find a match. In this case, we do: path.to[n] = path.to[n]

To make it to the output dictionary, there must be a match between the value path and the index path as described here.

The items in the lists must, at a minimum, have a .value property and a .full_path property

:param list value_matches:DatumInContext matches for the values :param list index_matches:DatumInContext matches for the indexes (used in the tuples)

Returns: Keys are the index’s value and values are the collection value
Return type: dict

parse_jpath(jpath, cache=None)

Takes the jsonpath given and compiles it into a jsonpath object that you can use

Parameters

jpath (str) – The jsonpath expression to be compiled
cache (dict) – (Optional) A dictionary to prevent running the compilation more than once. Compiling jsonpath is surprisingly expensive.

Returns

A parser object that wil match for the jsonpath expression given

Return type

callable

parse_payload(jpath, payload, cache=None)

Takes a jpath expression and a decoded json payload and returns the data from jsonpath_ng

The cache dictionary is used to prevent needing to re-compile jsonpath’s which is a very expensive operation.

Parameters

jpath (str) – A valid jpath expression
payload (object) – Generally a Dictionary with decoded json payload
cache (dict) – (Optional) Dictionary-compatible cache object of compiled parsers, indexed by the jsonpath.

Returns

jsonpath_ng.DatumInContext objects (they have .value and .fullpath attributes)

Return type

list

remove_trailing_suffix(path, separator='.', count=1)

Given something that is path-like, remove the trailing parts. For example, given: this.is.it, return this.is

Parameters

path (str) – The string to remove the tail of
separator (str) – (Optional) The separator to remove the end of (defaults to .)
count (int) – (Optional) How many trailing parts to remove)

Returns

String with the trailing parts that have been removed

Return type

str

Raises

ValueError – If you gave an invalid type for e.g. path

class silo.low_code_steps.standard.RegexParser

Bases: object

execute_regex(method_to_call, regex, data, flags)

Execute re.func function to get matched data.

Parameters

method_to_call (re.func) – re.function to execute.
regex (str) – valid regular expression.
data (str) – string where will be applied the regex.
flags (int) – sum of flags to take into account in the method execution.

Returns

re.function execution output.

Raises

ParserError – Not supported re.function on the parser.

generate_results(result)

Execute re.func function to get matched data.

generate_results(<re.Match object; span=(4, 7), match='o.C'>)
{
    "match": "o.C",
    "groups": (),
    "span": (4, 7)
 }

Parameters: result (object) – re.function execution output.
Returns: dictionary with relevant information.
Return type: dict

parse_arg_flags(step_config)

Parses the requested flags into a list

parse_arg_flags({"flags": "I,M"})
[re.IGNORECASE, re.MATCHALL]

Parameters: step_config (dict) – Configuration for the step
Returns: List of regex flags to use
Return type: list

parse_arg_method(step_config)

Validates and returns a callable re function.

parse_arg_flags({"method": "search"})
re.search

Parameters: step_config (dict) – Configuration for the step
Returns: callable function
Return type: re.function
Raises: ParserError – If the re.function is not callable.

parse_arg_regex(step_config)

Validates that the regex is valid

parse_arg_regex({"regex": "(.*))"}, "(.*)")
"(.*))"

parse_arg_regex({}, "(.*)")
"(.*))"

Parameters: step_config (dict) – Configuration for the step
Returns: Regular Expression to be executed
Return type: str
Raises: ParserError – Invalid regex or not provided.

parse_args(step_config)

Parse required args for the parser.

{
     "flags": ["I","M"],
     "method": "match",
     "regex": r"[a-z]\.[A-Z]")
 }
{
    "regex": r"[a-z]\.[A-Z]",
    "method": callable re.function,
    "flags": [re.IGNORECASE, re.MULTILINE]
 }

Parameters: step_config (dict) – Configuration for the step
Returns: parsed args in a dictionary.
Return type: dict

class silo.low_code_steps.standard.StaticValueRequest(credential, cache_read=True, result_cache=None)

Bases: RequestMaker

agent_supported = True

call(request, action_number)

Returns the first element in the requests

Parameters

request (ResultContainer) – ResultContainer that need to be executed
action_number (int) – Action number assigned to the execution of the action

Returns

ResultContainer where the result is value passed in

Return type

ResultContainer

static generate_agent_config(da_job, result_containers)

Generate an agent configuration for the Requestor / DA Job

This function generates all required configurations and returns them as a list of messages. This allows agent alignment to occur with the proper configuration to collect the required collections. If this function is not implemented, execution on the agent will not occur.

Parameters

da_job (dict) – Dynamic Application job pertaining to a specific app / device id.
result_containers (list) – List of result containers for the current collections

Returns

List of messages that can be used to generate the config

Return type

list

static get_desc()

static get_name()

static manipulate_agent_data(result)

Strip out the newline character

Parameters: result (object) – Results from the agent

validate_request(request)

Ensure the incoming ResultContainer is valid for the Requestor

Validate the ResultContainer is valid. If it is invalid, raise an exception to indicate the ResultContainer cannot be collected by the Requestor.

Parameters: request (ResultContainer) – Request to validate

class silo.low_code_steps.standard.YAMLArgParser(subs, rc)

Bases: ArgParser

SUPPORTED_VERSION = {'1.0': <function process_v1_0>, '2.0': <function process_v2_0>}

call(snippet_arg)

Convert a YAML snippet argument into a consumable format for the framework

The YAML object should contain a root element, low_code, and no other root elements. The sub-section, network, is required to have a usable format. The sub-section, processing, is optional. Each section defines steps in the following format:

Key / Value: The section contains one step
List: The section contains multiple steps

A step definition has the following requirements:

String: There is no configuration for the step [csv]
Key / Value (string) - The key is the step and the value is the action arg
[csv: type=dict]
Dictionary - There are multiple configuration options. The action argument will be
specified in the key “arg”, but is optional. Any other configuration will be specified by any other key / value pair.

An example would be as follows:

low_code:
  id: collect_id
  network:
      static_value:
        arg: '{"data": "value"},{"data2": "value2"},{"data3": "value3"}'
  processing:
      - csv
      - simple_key: 0
      - simple_key: 1
      - json
      - simple_key:
          cool: "beans"
          arg: "data2"

Parameters: snippet_arg (str) – Snippet arguments after substitution
Returns: Dictionary that instructs the framework on the order of steps to execute.
Return type: dict

static get_desc()

static get_name()

silo.low_code_steps.standard.cache_reader(result_container, action_arg)

OOTB Cache reader function

This step will read the previous step’s result from cache and return it.

Parameters

result_container – Device Metric for the current pipeline run
action_arg (dict) – Configuration parameters for the cache writer step

Returns

retreived cache entry that matches the provided cache_info

Return type

str

silo.low_code_steps.standard.cache_writer(result_container, action_arg, step_cache)

OOTB Cache Writer function

This step will write the previous step’s result to cache.

Parameters

result_container (ResultContainer) – Device Metric for the current pipeline run
action_arg (dict) – Configuration parameters for the cache writer step

Returns

None

silo.low_code_steps.standard.dummy(result): Return the value unchanged

silo.low_code_steps.standard.json_parser(result)

Converts the JSON string into a python dictionary

Parameters: result (object) – Result from the previous step
Returns: Python representation of the JSON object
Return type: object

silo.low_code_steps.standard.simple_key(result, action_arg)

For dicts, lists, etc, return the keys specified with periods separating.

For example: Path.to.value.

silo.low_code_steps.standard.store_data(action_arg, result_container)

silo.low_code_steps.standard.stringfloat(result)

Parse a string to a float.

If the string contains multiple floats, return the first one.

Raises: ResultsParsingError – When anything but a string with parsable digits is passed.

silo.low_code_steps.standard.stringint(result)

TODO: Filters out all characters and returns all numbers as a combined: integer. This is slightly strange behavior, because most people would not expect it to work that way, but it is retaining behavior from silo_ssh. Change this if it is not specifically needed.

Raises: ResultsParsingError – When a non-numeric string/value is passed.

silo.low_code_steps.standard package

Submodules