viper package

Submodules

viper.cli module

The Viper Command-line Interface

The Concept

After we define the tasks, actions, jobs etc. in the workspace, we need a way to execute them. Dropping into a Python shell and using the Python API is one way to do that. However, that might not be the most preferred approach for everyone.

Viper provides a command-line interface through the viper command to interact with the Python API without dropping into a Python shell.

Similarity Between the Python API and Command-line Interface

The CLI interface closely follows the Python API. Each of the subcommands with a colon (:) represents a method of a class or object. When we run viper --help, we can see the signatures of the methods/subcommands.

For example, the subcommand viper hosts:from-file represents the class method viper.Hosts.from_file(),

In the help menu the signature of this subcommand is defined as [Hosts] which means that it returns a text (JSON) representation of a Hosts, object which can be passed (piped) to another subcommand that expects the same via standard input.

On the other hand, the results:order-by has the signature [Results -> Results]. The subcommand represents the method viper.Results.order_by() and the signature [Results -> Results] means that the subcommand expects the text (JSON) representation of a Results object.

Example: Output Piping as Method Chaining

viper hosts:from-file("hosts.csv") \
        | viper hosts:task task.ping \
        | viper runners:run --max-workers 50 \
        | viper results:final \
        | viper results:order-by host.hostname host.ip \
        | viper results:to-file results.csv \
        | viper results:format "{host.hostname}: {stdout}"

In the above example, following things are happening:

  • The hosts:from-file subcommand with signature [Hosts] returns the text representation of a Hosts object.

  • hosts:task reads the output of hosts:from-file from standard input as it has the signature of [Hosts -> Runners] and returns Runners.

  • Then the runners:run subcommand with signature [Runners -> Results] reads the output of hosts:task from standard input and returns Results.

  • Finally results:format with signature [Results -> str] turns the Results into a string which cannot be passed (piped) to any further subcommand.

The data flow diagram:

hosts:from-file -> Hosts | hosts:task -> Runners | runners:run -> Results | results:final -> Results | results:order-by -> Results | results:to-file -> Results | results:format -> str

The above CLI example is equivalent to the following Python example:

from viper import Hosts
import task

print(
    Hosts.from_file("hosts.csv")
    .task(task.ping())
    .run(max_workers=50)
    .final()
    .order_by("host.hostname", "host.ip")
    .to_file("results.csv")
    .format("{host.hostname}: {stdout}")
)

Tip

Refer to Getting Started to see how task.ping and hosts.csv are written.

Defining Actions

Actions are simple Python functions that can be invoked using the viper lets subcommand.

Example:

Define an action in action.py:

cat > action.py << EOF

def add_them(a, b):
    return int(a) + int(b)

EOF

Now invoke the action:

viper lets action.add_them 5 10

Output:

15

Defining Viper Objects: Hosts, Task

Similar to actions, we can also define functions that return an instance of Task or Hosts. The *:from-func subcommands will invoke the function to get the object it returns.

Example: Define a host group in hosts.py

cat > hosts.py << EOF

from viper import Hosts, Host

def group1():
    return Hosts.from_items(
        Host("192.168.0.11", hostname="host11"),
        Host("192.168.0.12", hostname="host12"),
        Host("192.168.0.13", hostname="host13"),
        Host("192.168.0.14", hostname="host14"),
        Host("192.168.0.15", hostname="host15"),
    )

EOF

Get the hosts count in terminal:

viper hosts hosts.group1 | viper hosts:count

Output:

5

Note

viper hosts is an alias of viper hosts:from-func. Similarly, viper task is an alias of viper task:from-func.

However, viper results is an alias of viper results:from-history as there’s no reason to write results ourselves. It should come from the database.

So there’s no results:from-func, neither runners:from-func and so on.

Tip

Refer to Getting Started to find the example of task and job definitions.

Defining Utilities: Handlers, Filters, Sort Keys

Defining handlers, filters and sort keys are similar to defining actions but the first argument of the defined function is reserved for an instance of viper data type which it receives from the standard input.

Example:

Define a general handler in handler.py that operates on all Items instances:

cat > handler.py << EOF

import sys

def log_count(items, arg1):
    print(f"There are {items.count()} {arg1}.", file=sys.stderr)
    return items

EOF

Use the handler:

viper hosts hosts.group1 \
        | viper hosts:pipe handler.log_count hosts \
        | viper hosts:count

Output:

There are 5 hosts.
5

Note

Here arg1 recieves the second argument passed to hosts:pipe i.e. “hosts”.

Similarly filters and sort keys can be defined using functions having the first argument reserved for the object it will operate on, and the subsequent arguments for the variables that will be passed while invoking the *:filter and *:sort subcommands.

However, we hardly will need to really define filters and sort keys like this as most of the requirements of sorting and filtering should be satisfied with the *:order-by and *:where subcommands respectively.

viper.cli.func(objpath: str) → function[source]

Resolved Python function from given string.

viper.cli.run() → int[source]

viper.cli_base module

Base classes for the viper.cli module.

class viper.cli_base.SubCommand(subparser: argparse.ArgumentParser)[source]

Bases: object

Base class for the subcommand parsers.

add_arguments(parser: argparse.ArgumentParser) → None[source]
aliases = ()
classmethod attach_to(subparsers: argparse._SubParsersAction) → None[source]
name = None

viper.collections module

The Viper Python API

The Concept

Viper provides a powerful collection of data types such as Hosts, Runners, Results etc. and uses method chaining to perform different operations. The viper.collections module contains the collection of such data types. These data types share some common properties as all they inherit from the Collection class.

Example: Method Chaining

from viper import Hosts
import task

print(
    Hosts.from_file("hosts.csv")
    .task(task.ping())
    .run(max_workers=50)
    .final()
    .order_by("host.hostname", "host.ip")
    .to_file("results.csv")
    .format("{host.hostname}: {stdout}")
)

Tip

Refer to Getting Started to see how task.ping and hosts.csv are written.

Unit vs Container Types

The above mentioned data types can be categorised as unit and container types. The unit ones inherit from the Item class, while the container types inherit from Items class.

Below are the list of unit types and their container type counterparts:

Unit Types

Container Types

Task

Host

Hosts

Runner

Runners

Result

Results

Useful Common Properties & Abilities

The properties mentioned below are common to both unit and container type objects.

  • Immutable: All the datatypes are immutable i.e. they cannot be modified once initialized. This is to prevent any unexpected behaviour caused due to stateful-ness.

  • .from_json() and .to_json(): All the objects can be initialized from JSON texts using the .from_json() factory method and can be dumped back to JSON using the .to_json() method. This enables the objects to use a wide range of mediums such as the Unix pipes.

  • .format(): The objects can be converted to a string with a custof format using the .format() method.

    Example:

    host.format("{ip} {hostname} {meta.tag}")

Useful Abilities Common to the Unit Types

These abilities are common to Task, Host, Runner and Result unit type objects.

  • .from_dict() and .to_dict(): Helps representing the objects as Python dictionaries.

Useful Abilities Common to the Container Types

These abilities are common to Hosts, Runners and Results container type objects.

  • .from_items() and .to_items(): The .from_items() factory method is the recommended way to initialize container type objects. Although it can be a little slower, it removes duplicate items and performs other important checks before initializing the object. It supports sequences, generators, unit objects or all at once.

    Attention

    # Bad
Hosts((host1, host2, host3))

# Good
Hosts.from_items(host1, host2, host3)

    The .to_items() or the alias .all() returns the tuple of unit items back.

    Example:

    Hosts.from_items(
    host1, host2                      # Unit objects
    [host3, host4],                   # Sequence of objects
    (host for host in list_of_hosts)  # Generator of objects
).to_items()

  • .from_file() and .to_file(): Container type objects can be initialized from text files and dumped back to text files with certain formats (currently supported json, yml and csv) using these methods.

    Example:

    Hosts.from_file("hosts.json").to_file("hosts.csv")

  • .from_list() and .to_list(): Similar to unit types’ .from_dict() and .to_dict() but operates with list of dictionaries that represent the unit type objects.

  • .count(): Returns the count of items it holds.

  • .head() and .tail(): Returns an instance of the same container type object containing first or last n items (n defaults to 10).

    Example:

    # Get the set of last 5 items from the set of first 10 items.
hosts.head(10).tail(5)

  • .range(): Similar to .head() or .tail() but enables us to define a range (like Python’s list[i:j] indexing).

    Example:

    # Exclude the last item (like like Python's list[0:-1])
hosts.range(0, -1)

  • .sort(): Similar to Python’s list.sort() but returns a new instance instead of making changes to the existing object (which is impossible because of immutability).

    Example:

    # Reverse sort by IP, then by hostname
hosts.sort(key=lambda host: [host.ip, host.hostname], reverse=True)

  • .order_by(): Similar to .sort() but expects the field names instead of a function. Inspired by SQL.

    Example:

    # Reverse sort by ip, then by hostname
hosts.order_by("ip", "hostname", reverse=True)

  • .filter(): Similar to Python’s filter() but returns an instance of the same container type object containing the filtered items.

    Example:

    # Filter hosts where hostname starts with "foo"
hosts.filter(lambda host: host.hostname.startswith("foo"))

  • .where(): Similar to filter, but expects the field name, the condition and the value instead of a function. Inspired by SQL.

    Example:

    # Filter hosts where the hostname starts with "foo"
hosts.where(
    "hostname", WhereConditions.startswith, ["foo"]
)

More on Task: Command Factories, Output Processors, Callbacks and …

The minimum requirements of defining a Task is to pass the task name and the command factory. Optionally, we can also pass the stdout and stderr processors, and also the pre and post run callbacks.

The command factory expects a Host object and returns a tuple of string.

Example:

def ping_command(host):
    return "ping", "-c", "1", host.ip

The stdout and stderr processors expect a string and return a string.

Example:

def strip_output(txt):
    return txt.strip()

The pre run callback expects a Runner object and doesn’t return anything. While the post run callback expects a Result object and doesn’t return anything either.

Example:

import sys

def log_command_pre_run(runner):
    command = runner.task.command_factory(runner.host, *runner.args)
    print("Running command:", command, file=sys.stderr)

def log_result_post_run(result):
    print("OK:" if result.ok() else "ERROR:", result.host.hostname, file=sys.stderr)

Note

Logs are being printed to stderr as stdout is for the JSON encoded Results object.

Attention

The arguments command_factory, stdout_processor, stderr_processor, pre_run and post_run callbacks expect normal functions, not lambdas.

# Bad
def ping():
    return Task(
        name="Ping once",
        command_factory=lambda host: "ping", "-c", "1", host.ip,
        stdout_processor=lambda txt: txt.strip(),
        stderr_processor=lambda txt: txt.strip(),
        pre_run=lambda runner: print(runner.to_dict(), file=sys.stderr),
        post_run=lambda result: print(result.to_dict(), file=sys.stderr),
    )

# Good
def ping():
    return Task(
        name="Ping once",
        command_factory=ping_command,
        stdout_processor=strip_output,
        stderr_processor=strip_output,
        pre_run=log_command_pre_run,
        post_run=log_result_post_run,
    )

Apart from these, a Task also optionally expects timeout, retry and meta.

  • timeout: The execution will timeout after the specified seconds if timeout is defined.

    The countdown doesn’t count the time spent on the pre and post run callbacks, neither the command factory invocation. It only counts time spent on executing the generated command.

  • retry: It defaults to 0. If more than 0, The runner will re-invoke the run() method with the updated retry value if the command execution fails. The results generated for these retries will be stored in DB and will be available in history. They will have the same trigger_time but different start and end time values.

    However, if the failure is caused by any reason other than the actual command execution, such as while invoking the command factory or output processors or pre/post run callbacks, a Python error will be raised which won’t be stored in DB. If any such error occurs while running the task in batch, it will be ignored with the traceback printed to stderr.

  • meta: It is the same as the meta field in Host. The value should be generated only using the viper.meta() function.

    Attention

    # Bad
def ping():
    return Task(
        name="Ping once",
        command_factory=ping_command,
        meta={"tag": "foo"},
    )

# Good
def ping():
    return Task(
        name="Ping once",
        command_factory=ping_command,
        meta=meta(tag="foo")
    )
class viper.collections.WhereConditions[source]

Bases: enum.Enum

Where query conditions for viper Items.

Example

Hosts.from_items(
    Hosts('1.1.1.1'),
    Hosts('2.2.2.2'),
).where("ip", WhereConditions.is_, '1.1.1.1')
contains = 'CONTAINS'
endswith = 'ENDSWITH'
is_ = 'IS'
is_not = 'IS_NOT'
not_contains = 'NOT_CONTAINS'
not_endswith = 'NOT_ENDSWITH'
not_startswith = 'NOT_STARTSWITH'
startswith = 'STARTSWITH'
class viper.collections.Collection[source]

Bases: object

The base collection class.

This is the parent class for all Viper native objects such as viper.collections.Host, viper.collections.Results etc.

classmethod from_func(funcpath: str) → CollectionType[source]

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → CollectionType[source]

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int[source]

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

pipe(handler: Any, *args: Any) → T[source]

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str[source]

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
viper.collections.meta(**mapping: Union[str, bool, int, float, None]) → Any[source]

Meta data object creator.

Parameters

**mapping – key-value pairs as the meta data.

Example

from viper import Host, meta

host = Host('1.1.1.1', meta=meta(provider="aws"))
host.meta.provider
# 'aws'
host.meta['provider']
# 'aws'
host['meta']['provider']
# 'aws'
class viper.collections.Item[source]

Bases: viper.collections.Collection

The base class for a single Viper object.

viper.collections.Host, viper.collections.Task etc. inherits this base class.

format(template: str) → str[source]

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → ItemType[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType[source]

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
to_dict(*, flatten: bool = False) → Dict[str, Any][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str[source]

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.collections.Items(_all: Sequence[T] = <factory>)[source]

Bases: viper.collections.Collection, typing.Generic

The base class for collection objects for a group of similar items.

viper.collections.Hosts, viper.collections.Results etc. inherits this base class.

all() → Sequence[T][source]

Alias for .to_items()

count() → int[source]

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType[source]

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

format(template: str, sep: str = '\n') → str[source]

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType[source]

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType[source]

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType[source]

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType[source]

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType[source]

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

order_by(*properties: str, reverse: bool = False) → ItemsType[source]

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType[source]

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType[source]

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType[source]

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

to_file(filepath: str) → ItemsType[source]

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T][source]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str[source]

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]][source]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType[source]

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])
class viper.collections.Host(ip: str, hostname: Optional[str] = None, domain: Optional[str] = None, port: int = 22, login_name: Optional[str] = None, identity_file: Optional[str] = None, meta: Any = <factory>)[source]

Bases: viper.collections.Item

Viper Host class.

domain = None
format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
fqdn() → str[source]

Get the FQDN from hostname and domain name.

Return type

str

Raises

ValueError – If either of hostname or domain name is not set.

classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Host[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

hostname = None
identity_file = None
login_name = None
pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
port = 22
results() → viper.collections.Results[source]

Fetch recent results of current host from database.

run_task(task: viper.collections.Task, *args: str) → viper.collections.Result[source]

Assign the task to the host and then run it.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from viper.collections.Task.command_factory.

Return type

viper.collections.Result

Example

Host("1.2.3.4").task_task(ping)
task(task: viper.collections.Task, *args: str) → viper.collections.Runner[source]

Assigns a task to be run.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from command_factory.

Return type

viper.collections.Runner

Example

Host("1.2.3.4").task(ping)
to_dict(*, flatten: bool = False) → Dict[str, object][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.collections.Hosts(_all: Sequence[viper.collections.Host] = <factory>)[source]

Bases: viper.collections.Items

A group of viper.collections.Host objects.

all() → Sequence[T]

Alias for .to_items()

count() → int

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

format(template: str, sep: str = '\n') → str

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

order_by(*properties: str, reverse: bool = False) → ItemsType

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

results() → viper.collections.Results[source]

Get the past results of this group of hosts from history.

Return type

viper.collections.Results

run_task(task: viper.collections.Task, *args: str, max_workers: int = 0) → viper.collections.Results[source]

Assign the task to the host and then run it.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from viper.collections.Task.command_factory.

  • max_workers (int) – Maximum number of thread workers. if the value is <= 1, tasks will run in sequence.

Return type

viper.collections.Results

Example

Hosts.from_items(Host("1.2.3.4")).task_task(ping)
sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

task(task: viper.collections.Task, *args: str) → viper.collections.Runners[source]

Assigns a task to be run on each host in the group.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from command_factory.

Return type

viper.collections.Runners

Example

Hosts.from_items(Host("1.2.3.4")).task(ping)
to_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])
class viper.collections.Task(name: str, command_factory: Any, timeout: Union[int, float, None] = None, retry: int = 0, stdout_processor: Optional[Callable[[str], str]] = None, stderr_processor: Optional[Callable[[str], str]] = None, pre_run: Optional[Callable[[viper.collections.Runner], None]] = None, post_run: Optional[Callable[[viper.collections.Result], None]] = None, meta: Any = <factory>)[source]

Bases: viper.collections.Item

A task is a definition of a single operation.

A task must contain name and a named function to generate the command.

format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Task[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
post_run = None
pre_run = None
results() → viper.collections.Results[source]

Get the past results of this task.

Return type

Rviper.collections.esults

retry = 0
stderr_processor = None
stdout_processor = None
timeout = None
to_dict(*, flatten: bool = False) → Dict[str, object][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.collections.Runner(host: viper.collections.Host, task: viper.collections.Task, args: Tuple[str, ...] = ())[source]

Bases: viper.collections.Item

A single task runner.

A runner owns the responsibility of actually generating and running the command, store the result, calling, retrying, etc. We generally do not use it directly.

args = ()
format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Runner[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
run(retry: int = 0, trigger_time: Optional[float] = None) → viper.collections.Result[source]

Run the task on the host.

Parameters

  • retry (int) – Count of retries used.

  • trigger_time (optional) (float) – The trigger time used for grouping (auto generated).

Return type

viper.collections.Result

to_dict(*, flatten: bool = False) → Dict[str, Any][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.collections.Runners(*args, **kwds)[source]

Bases: viper.collections.Items

all() → Sequence[T]

Alias for .to_items()

count() → int

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

format(template: str, sep: str = '\n') → str

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

hosts() → viper.collections.Hosts[source]

Get the list of hosts from the runners.

Return type

viper.collections.Hosts

order_by(*properties: str, reverse: bool = False) → ItemsType

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

run(max_workers: int = 0) → viper.collections.Results[source]

Run the tasks.

Parameters

max_workers (int) – Maximum number of thread workers to use.

Return type

viper.collections.Results

sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

to_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])
class viper.collections.Result(trigger_time: float, task: viper.collections.Task, host: viper.collections.Host, args: Tuple[str, ...], command: Tuple[str, ...], stdout: str, stderr: str, returncode: int, start: float, end: float, retry: int)[source]

Bases: viper.collections.Item

The result of an executed task.

classmethod by_id(id_: int) → viper.collections.Result[source]

Fetch the result from DB by hash.

Parameters

hash (int) – The hash value.

Return type

viper.collections.Result

errored() → bool[source]

returns True if the result is failure.

Return type

bool

format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Result[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

ok() → bool[source]

Returns True the result is success.

Return type

bool

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
retry_left() → int[source]

Get how many retries are left.

Return type

int

runner() → viper.collections.Runner[source]

Recreate the runner from the result.

Return type

viper.collections.Runner

save() → viper.collections.Result[source]

Save the result in DB.

Return type

viper.collections.Result

to_dict(*, flatten: bool = False) → Dict[str, Any][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.collections.Results(_all: Sequence[viper.collections.Result] = <factory>)[source]

Bases: viper.collections.Items

A group of viper.collections.Results.

all() → Sequence[T]

Alias for .to_items()

classmethod by_host(host: viper.collections.Host) → viper.collections.Results[source]

Fetch and return results from history of the given host.

Parameters

host (viper.collections.Host) – Fetch results of this host.

Return type

viper.collections.Results

classmethod by_task(task: viper.collections.Task) → viper.collections.Results[source]

Fetch and return results from history of the given task.

Parameters

host (viper.collections.Host) – Fetch results of this task.

Return type

viper.collections.Results

count() → int

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

final() → viper.collections.Results[source]

Get the final results only (ignoring the previous retries).

Return type

viper.collections.Results

format(template: str, sep: str = '\n') → str

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_history(final: bool = False) → viper.collections.Results[source]

Fetch and return all the results from history.

Return type

ciper.collections.Results

classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

hosts() → viper.collections.Hosts[source]

Get the list of hosts from the results.

Return type

viper.collections.Hosts

order_by(*properties: str, reverse: bool = False) → ItemsType

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

re_run(max_workers: int = 0) → viper.collections.Results[source]

Recreate the runners from the results and run again.

Return type

viper.collections.Results

runners() → viper.collections.Runners[source]

Recreate the runners from the results.

Return type

viper.collections.Runners

sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

to_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])

viper.const module

Common contants are defined here.

class viper.const.Config[source]

Bases: enum.Enum

Default viper configuration.

db_url = 'viperdb.sqlite3'
max_workers = 0
modules_path = '.'

viper.db module

class viper.db.ViperDB(url: 'str' = 'viperdb.sqlite3')[source]

Bases: object

engine = None
classmethod init(url: str, force: bool = False) → None[source]
url = 'viperdb.sqlite3'

viper.main module

viper.main.main() → None[source]

The main entrypoint to Viper CLI.

viper.project module

Extending the Command-line Interface (using viperfile.py)

Why and How

The viper CLI can easily be extended to include custom subcommands using the viper.project module.

To do this, you have to create a file named viperfile.py in the root of your workspace. This file will contain the definition(s) of one or multiple projects. A project works like a namespace for all the custom subcommands under it.

Example: Defining a Project

This is how a project can be defined in viperfile.py:

from viper.project import Project, arg

foo = Project("foo")

The viper.project.arg() function helps defining the command-line arguments a.k.a options or switches that the subcommand expects.

Let’s define a subcommand @foo:group1 that expects optional arguments --login_name and --identity_file with some default values and returns the text representation of a viper.Hosts object.

Example: Defining a subcommand for host group

from viper import Host, Hosts, meta

@foo.hostgroup(
    args=[
        arg("-l", "--login_name", default="root"),
        arg("-i", "--identity_file", default="/root/.ssh/id_rsa.pub"),
    ]
)
def group1(args):
    return Hosts.from_items(
        Host(
            ip="192.168.0.11",
            hostname="host11"
            login_name="root",
            identity_file=args.identity_file,
            meta=meta(provider="aws"),
        ),
        Host(
            ip="192.168.0.12",
            hostname="host12",
            login_name="root",
            identity_file=args.identity_file,
            meta=meta(provider="aws"),
        )
    )

Now running viper -h in that workspace will show us @foo:group1  [Hosts], and running viper @foo:group1 --help will list the arguments it’s expecting and their default values.

The subcommand can now be executed as below:

# Use the default values
viper @foo:group1

# Specify the login name and identity file
viper @foo:group1 -l user1 -i ~user1/.ssh/id_rsa.pub

Note

All the custom subcommands are prefixed with @ to separate them from the core viper subcommands. And the string following @ acts like a namespace that separates the subcommands belonging from different projects in the same viperfile.

class viper.project.Project(prefix: str, action_commands: List[Type[viper.cli_base.SubCommand]] = <factory>, hostgroup_commands: List[Type[viper.cli_base.SubCommand]] = <factory>, filter_commands: List[Type[viper.cli_base.SubCommand]] = <factory>, handler_commands: List[Type[viper.cli_base.SubCommand]] = <factory>, job_commands: List[Type[viper.cli_base.SubCommand]] = <factory>)[source]

Bases: object

A project (namespace) for the viper sub commands

Parameters

prefix (str) – Prefix for the related sub commands.

When we define a project, we basically define a namespace (a prefix) for the commands.

action(args: Optional[Sequence[Tuple[Tuple[str, ...], Dict[str, Any]]]] = None, totype: Optional[type] = None) → Callable[[Callable[[argparse.Namespace], T]], Callable[[argparse.Namespace], T]][source]

Use this decorator to define an action.

Parameters

args (optional) (list) – List of arguments for argparse.ArgumentParser.

all_commands() → List[Type[viper.cli_base.SubCommand]][source]

Return all sub commands

handler(fromtype: type, totype: type, args: Optional[Sequence[Tuple[Tuple[str, ...], Dict[str, Any]]]] = None) → Callable[[Callable[[T, argparse.Namespace], C]], Callable[[T, argparse.Namespace], C]][source]

Use this decorator to define handlers

Parameters

  • fromtype (type) – The type of object this handler is expecting.

  • totype (type) – The type of object this handler returns.

  • args (optional) (list) – List of arguments for argparse.ArgumentParser.

hostgroup(args: Optional[Sequence[Tuple[Tuple[str, ...], Dict[str, Any]]]] = None) → Callable[[Callable[[argparse.Namespace], viper.collections.Hosts]], Callable[[argparse.Namespace], viper.collections.Hosts]][source]

Use this decorator to define host groups

Parameters

args (optional) (list) – Arguments to be parsed by py:class:argparse.ArgumentParser

job(args: Optional[Sequence[Tuple[Tuple[str, ...], Dict[str, Any]]]] = None) → Callable[[Callable[[viper.collections.Hosts, argparse.Namespace], viper.collections.Results]], Callable[[viper.collections.Hosts, argparse.Namespace], viper.collections.Results]][source]

Use this decorator to define a job.

Parameters

args (optional) (list) – List of arguments for argparse.ArgumentParser.

viper.project.arg(*args: str, **kwargs: Any) → Tuple[Tuple[str, ...], Dict[str, Any]][source]

Argumenst to be passed to argparse

viper.serializers module

Viper Object Serializers

This module contains various loaders that helps loading viper collection objects (such as Hosts, Task etc.) from a file, and also dumpers that helps dump tha same to a file.

class viper.serializers.Serializers[source]

Bases: enum.Enum

A list of serializers to help loading and dumping objects from or to a file.

csv = Serializer(load=<function _load_csv>, dump=<function _dump_csv>)
json = Serializer(load=<function _load_json>, dump=<function _dump_json>)
yml = Serializer(load=<function _load_yaml>, dump=<function _dump_yaml>)

viper.utils module

A collection of some handy utilities.

viper.utils.flatten_dict(d: Dict[Any, Any]) → Dict[str, object][source]

Flatten a nested dictionary.

Parameters

d (dist) – The nested dictionary to flatten.

Return type

dict

Example

flatten_dict({"a": 1, "b": {"c": 2, "d": {"e": 3, "f": 4}}})
# {
#     "a": 1,
#     "b:c": 2,
#     "b:d:e": 3,
#     "b:d:f": 4,
# }
viper.utils.optional(dict_: Dict[object, object], key: object, expect: Union[type, Tuple[type, ...]], *, parser: Optional[Callable[[Dict[object, object], object], object]] = None) → Optional[T][source]

Get a value safely from the given dict using the given key. If the value is not found or is None, return it anyway.

Raises

ValueError

viper.utils.required(dict_: Dict[object, object], key: object, expect: Union[type, Tuple[type, ...]], *, parser: Optional[Callable[[Dict[object, object], object], object]] = None, default: Optional[T] = None, default_factory: Optional[Callable[[], T]] = None) → T[source]

Get a value safely from the given dict using the given key. If the value is not found or is None, raise value error.

Raises

ValueError

viper.utils.unflatten_dict(d: Dict[object, object]) → Dict[object, object][source]

Convert a flattened dict back to nested dict. :param dict d: The flattened dict to convert. :rtype: dict :example:

unflatten_dict({"a": 1, "b:c": 2, "b:d:e": 3, "b:d:f": 4})
# {
#   "a": 1,
#   "b": {"c": 2, "d": {"e": 3, "f": 4}},
# }

Module contents

Viper provides a simple API (Python and CLI) to easily manage large infrastructures.

    ▄   ▄█ █ ▄▄  ▄███▄   █▄▄▄▄   ▄█    ▄   ▄████  █▄▄▄▄ ██     ▄█▄    ████▄ █▀▄▀█ █▀▄▀█ ██      ▄   ██▄   ▄███▄   █▄▄▄▄
     █  ██ █   █ █▀   ▀  █  ▄▀   ██     █  █▀   ▀ █  ▄▀ █ █    █▀ ▀▄  █   █ █ █ █ █ █ █ █ █      █  █  █  █▀   ▀  █  ▄▀
█     █ ██ █▀▀▀  ██▄▄    █▀▀▌    ██ ██   █ █▀▀    █▀▀▌  █▄▄█   █   ▀  █   █ █ ▄ █ █ ▄ █ █▄▄█ ██   █ █   █ ██▄▄    █▀▀▌
 █    █ ▐█ █     █▄   ▄▀ █  █    ▐█ █ █  █ █      █  █  █  █   █▄  ▄▀ ▀████ █   █ █   █ █  █ █ █  █ █  █  █▄   ▄▀ █  █
  █  █   ▐  █    ▀███▀     █      ▐ █  █ █  █       █      █   ▀███▀           █     █     █ █  █ █ ███▀  ▀███▀     █
   █▐        ▀            ▀         █   ██   ▀     ▀      █                   ▀     ▀     █  █   ██                ▀
   ▐                                                     ▀                               ▀

Getting Started

Installation

pip install -U viper-infra-commander

# Or install with batteries included

pip install -U "viper-infra-commander[batteries]"

Initialization

Viper needs to initialize a SQLite DB in the present working directory (the workspace).

# (Optional) enable tab auto completion
eval "$(viper autocomplete $(basename $SHELL))"


# Initialize SQLite DB
viper init -f

Viper in Action (Basic Mode)

Define a set of hosts in csv format (json and yml are also supported) in hosts.csv:

cat > hosts.csv << EOF

ip,hostname,login_name,identity_file
192.168.0.11,host11,root,/root/.ssh/id_rsa.pub
192.168.0.12,host12,root,/root/.ssh/id_rsa.pub
192.168.0.13,host13,root,/root/.ssh/id_rsa.pub
192.168.0.14,host14,root,/root/.ssh/id_rsa.pub
192.168.0.15,host15,root,/root/.ssh/id_rsa.pub

EOF

Define a task in task.py:

cat > task.py << EOF

from viper import Task

def ping_command(host):
    return "ping", "-c", "1", host.ip

def ping():
    return Task(
        name="Ping once",
        command_factory=ping_command
    )

EOF

Perform the following actions:

  • Run the ping task on the set of hosts in parallel with 5 workers

  • Then filter only the results where the task failed

  • Re-run the task on the filtered set of hosts

  • Store all the results

viper hosts:from-file hosts.csv \
        | viper hosts:run-task task.ping --max-worker 5 \
        | viper results:where returncode IS_NOT 0 \
        | viper results:re-run --indent 4

The results are stored in the DB as history. To see the stdout of the final results from history:

viper results \
        | viper results:final \
        | viper results:format "{host.hostname}: {stdout}"

Export the results to a csv file:

viper results --final \
        | viper results:to-file results.csv --indent 4

Define a job using the Python API (CLI and Python API are almost similar):

cat > job.py << EOF

from viper import WhereConditions
from task import ping

def ping_and_export(hosts):
    return (
        hosts.task(ping())
        .run(max_workers=5)
        .final()
        .to_file("results.csv")
    )

EOF

Run the job using CLI:

viper hosts:from-file hosts.csv \
        | viper run job.ping_and_export \
        | viper results:format "{host.hostname}: {stdout}"

Viperfile in Action (Advanced Mode)

Define a project with custom subcommands in viperfile:

cat > viperfile.py << EOF

from viper import Hosts, Task
from viper.project import Project, arg


foo = Project(prefix="foo")


@foo.hostgroup(args=[arg("-f", "--file", default="hosts.csv")])
def allhosts(args):
    return Hosts.from_file(args.file)


def remote_exec_command(host, command):
    return (
        "ssh",
        "-i",
        host.identity_file,
        "-l",
        host.login_name,
        "-p",
        str(host.port),
        "-o",
        "StrictHostKeyChecking=no",
        "-o",
        "PubkeyAuthentication=yes",
        host.ip,
        command,
    )


@foo.job(
    args=[
        arg("command", help="command to execute"),
        arg("-w", "--workers", type=int, default=1),
    ]
)
def remote_exec(hosts, args):
    return (
        hosts.task(
            Task(
                name="Remote execute command",
                command_factory=remote_exec_command,
                timeout=5,
            ),
            args.command,
        )
        .run(max_workers=args.workers)
        .final()
    )

EOF

See the auto generated custom subcommands:

viper --help
# Will show the subcommands: "@foo:allhosts" and "@foo:remote_exec"

Run the job, export the results in a file, and format the output:

viper @foo:allhosts \
        | viper @foo:remote_exec "uname -a" --workers 5 \
        | viper results:to-file results.csv \
        | viper results:format "{task.name} [{host.hostname}]: {returncode}: {stdout}"
viper.meta(**mapping: Union[str, bool, int, float, None]) → Any[source]

Meta data object creator.

Parameters

**mapping – key-value pairs as the meta data.

Example

from viper import Host, meta

host = Host('1.1.1.1', meta=meta(provider="aws"))
host.meta.provider
# 'aws'
host.meta['provider']
# 'aws'
host['meta']['provider']
# 'aws'
class viper.Host(ip: str, hostname: Optional[str] = None, domain: Optional[str] = None, port: int = 22, login_name: Optional[str] = None, identity_file: Optional[str] = None, meta: Any = <factory>)[source]

Bases: viper.collections.Item

Viper Host class.

domain = None
format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
fqdn() → str[source]

Get the FQDN from hostname and domain name.

Return type

str

Raises

ValueError – If either of hostname or domain name is not set.

classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Host[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

hostname = None
identity_file = None
login_name = None
pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
port = 22
results() → viper.collections.Results[source]

Fetch recent results of current host from database.

run_task(task: viper.collections.Task, *args: str) → viper.collections.Result[source]

Assign the task to the host and then run it.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from viper.collections.Task.command_factory.

Return type

viper.collections.Result

Example

Host("1.2.3.4").task_task(ping)
task(task: viper.collections.Task, *args: str) → viper.collections.Runner[source]

Assigns a task to be run.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from command_factory.

Return type

viper.collections.Runner

Example

Host("1.2.3.4").task(ping)
to_dict(*, flatten: bool = False) → Dict[str, object][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.Hosts(_all: Sequence[viper.collections.Host] = <factory>)[source]

Bases: viper.collections.Items

A group of viper.collections.Host objects.

all() → Sequence[T]

Alias for .to_items()

count() → int

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

format(template: str, sep: str = '\n') → str

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

order_by(*properties: str, reverse: bool = False) → ItemsType

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

results() → viper.collections.Results[source]

Get the past results of this group of hosts from history.

Return type

viper.collections.Results

run_task(task: viper.collections.Task, *args: str, max_workers: int = 0) → viper.collections.Results[source]

Assign the task to the host and then run it.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from viper.collections.Task.command_factory.

  • max_workers (int) – Maximum number of thread workers. if the value is <= 1, tasks will run in sequence.

Return type

viper.collections.Results

Example

Hosts.from_items(Host("1.2.3.4")).task_task(ping)
sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

task(task: viper.collections.Task, *args: str) → viper.collections.Runners[source]

Assigns a task to be run on each host in the group.

Parameters

  • task (viper.collections.Task) – The task to be assigned.

  • args (str) – The arguments to be used to create the command from command_factory.

Return type

viper.collections.Runners

Example

Hosts.from_items(Host("1.2.3.4")).task(ping)
to_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])
class viper.Result(trigger_time: float, task: viper.collections.Task, host: viper.collections.Host, args: Tuple[str, ...], command: Tuple[str, ...], stdout: str, stderr: str, returncode: int, start: float, end: float, retry: int)[source]

Bases: viper.collections.Item

The result of an executed task.

classmethod by_id(id_: int) → viper.collections.Result[source]

Fetch the result from DB by hash.

Parameters

hash (int) – The hash value.

Return type

viper.collections.Result

errored() → bool[source]

returns True if the result is failure.

Return type

bool

format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Result[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

ok() → bool[source]

Returns True the result is success.

Return type

bool

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
retry_left() → int[source]

Get how many retries are left.

Return type

int

runner() → viper.collections.Runner[source]

Recreate the runner from the result.

Return type

viper.collections.Runner

save() → viper.collections.Result[source]

Save the result in DB.

Return type

viper.collections.Result

to_dict(*, flatten: bool = False) → Dict[str, Any][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.Results(_all: Sequence[viper.collections.Result] = <factory>)[source]

Bases: viper.collections.Items

A group of viper.collections.Results.

all() → Sequence[T]

Alias for .to_items()

classmethod by_host(host: viper.collections.Host) → viper.collections.Results[source]

Fetch and return results from history of the given host.

Parameters

host (viper.collections.Host) – Fetch results of this host.

Return type

viper.collections.Results

classmethod by_task(task: viper.collections.Task) → viper.collections.Results[source]

Fetch and return results from history of the given task.

Parameters

host (viper.collections.Host) – Fetch results of this task.

Return type

viper.collections.Results

count() → int

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

final() → viper.collections.Results[source]

Get the final results only (ignoring the previous retries).

Return type

viper.collections.Results

format(template: str, sep: str = '\n') → str

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_history(final: bool = False) → viper.collections.Results[source]

Fetch and return all the results from history.

Return type

ciper.collections.Results

classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

hosts() → viper.collections.Hosts[source]

Get the list of hosts from the results.

Return type

viper.collections.Hosts

order_by(*properties: str, reverse: bool = False) → ItemsType

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

re_run(max_workers: int = 0) → viper.collections.Results[source]

Recreate the runners from the results and run again.

Return type

viper.collections.Results

runners() → viper.collections.Runners[source]

Recreate the runners from the results.

Return type

viper.collections.Runners

sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

to_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])
class viper.Runner(host: viper.collections.Host, task: viper.collections.Task, args: Tuple[str, ...] = ())[source]

Bases: viper.collections.Item

A single task runner.

A runner owns the responsibility of actually generating and running the command, store the result, calling, retrying, etc. We generally do not use it directly.

args = ()
format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Runner[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
run(retry: int = 0, trigger_time: Optional[float] = None) → viper.collections.Result[source]

Run the task on the host.

Parameters

  • retry (int) – Count of retries used.

  • trigger_time (optional) (float) – The trigger time used for grouping (auto generated).

Return type

viper.collections.Result

to_dict(*, flatten: bool = False) → Dict[str, Any][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.Runners(*args, **kwds)[source]

Bases: viper.collections.Items

all() → Sequence[T]

Alias for .to_items()

count() → int

Count the number of items.

Return type

int

filter(filter_: Any, *args: object) → ItemsType

Filter the items by a given function.

Parameters

  • filter (callable) – A function that returns either True or False.

  • args (object) – Arguments to be passed to the filter to manipulate the decision making.

format(template: str, sep: str = '\n') → str

Get a custom string representation of this list of objects.

Parameters

  • template (str) – The template will be compiled by Python’s .format().

  • sep (str) – The separator used to separate all the items.

Return type

str

Example

Hosts.from_items(Host("1.2.3.4")).format("{ip} {hostname} {meta.tag}")
classmethod from_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to read from.

Return type

viper.collections.Items

Example

Hosts.from_file("/path/to/file/hosts.json")

Hosts.from_file("/path/to/file/hosts.csv")

Hosts.from_file("/path/to/file/hosts.yml")
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_items(*items: Union[Iterable[T], T]) → ItemsType

Create this an instance of this object using the given items.

Parameters

items (viper.collections.Item) – The group of items to hold.

Return type

viper.collections.Items

Note

Classes that inherits from Items should not be initialized directly (e.g. Hosts(Host("1.2.3.4"))). Factory methods such as this should be used instead.

Example

Hosts.from_items(Host("1.2.3.4"))
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemsType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
classmethod from_list(list_: Sequence[Dict[object, object]], *, unflatten: bool = False) → ItemsType

Initialize items from given list of dictiionaries.

Parameters

list (list) – The list of dictiionaries containing the item properties.

Return type

viper.collections.Items

This is used for loading JSON data into an instance of this class.

Example

Hosts.from_list([{"ip": "1.2.3.4"}])
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

head(n: int = 10) → ItemsType

Get the first ‘n’ items from the group of items.

Works like Python’s [:n] index

Return type

viper.collections.Item

hosts() → viper.collections.Hosts[source]

Get the list of hosts from the runners.

Return type

viper.collections.Hosts

order_by(*properties: str, reverse: bool = False) → ItemsType

Sort the items by multiple properties

Parameters

  • properties (list) – The item will be sorted by the given properties in order.

  • reverse (bool) – Reverse the order after sort.

Return type

Items

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
range(i: Optional[int] = None, j: Optional[int] = None) → ItemsType

get the items in given range.

Parameters

  • i (optional) (int) – The first index.

  • j (optional) (int) – The last index.

Return type

Items

This has the same behaviour as Python’s [i:j]

run(max_workers: int = 0) → viper.collections.Results[source]

Run the tasks.

Parameters

max_workers (int) – Maximum number of thread workers to use.

Return type

viper.collections.Results

sort(key: Optional[Callable[[viper.collections.Item], object]] = None, reverse: bool = False) → ItemsType

Sort the items by given key/function.

Parameters

  • key (callable) – A function similar to the one passed to the built-in sorted().

  • reverse (bool) – Reverse the order after sort.

Return type

Items

tail(n: int = 10) → ItemsType

Get the last ‘n’ item from the group of items.

Works like Python’s [-n:] index

Return type

viper.collections.Item

to_file(filepath: str) → ItemsType

Initialize items by reading data from a file.

Parameters

str (filepath) – The path for the file to write.

Return type

viper.collections.Items

Example

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.json")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.csv")

Hosts.from_items(Host("1.2.3.4")).to_file("/path/to/file/hosts.yml")
to_items() → Sequence[T]

Get a tuple of all the items.

Return type

tuple

to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
to_list(flatten: bool = False) → List[Dict[str, object]]

Represent the items as a list of dictiionaries.

This is used for dumping an instance of this object as JSON data.

Parameters

flatten (bool) – If True, dicts won’t be nested.

Return type

list

Example

Hosts.from_items(Host("1.2.3.4")).to_list()
where(key: str, condition: viper.collections.WhereConditions, values: Sequence[str]) → ItemsType

Select items by a custom query.

Parameters

  • key (str) – The key will be compiled by Python’s .format().

  • condition (viper.collections.WhereConditions) – The where condition.

  • values (list) – The values for the key.

Return type

Items

Example

Hosts.from_items(
    Host("1.1.1.1"),
    Host("2.2.2.2")
).where("ip", WhereConditions.is_, ["1.1.1.1"])
class viper.Task(name: str, command_factory: Any, timeout: Union[int, float, None] = None, retry: int = 0, stdout_processor: Optional[Callable[[str], str]] = None, stderr_processor: Optional[Callable[[str], str]] = None, pre_run: Optional[Callable[[viper.collections.Runner], None]] = None, post_run: Optional[Callable[[viper.collections.Result], None]] = None, meta: Any = <factory>)[source]

Bases: viper.collections.Item

A task is a definition of a single operation.

A task must contain name and a named function to generate the command.

format(template: str) → str

Get a custom string representation of this object.

Parameters

template (str) – The template to be used as “template”.format(**vars(self))

Return type

str

Example

Host("1.2.3.4").format("{ip} {hostname} {meta.tag}")
classmethod from_dict(dict_: Dict[object, object], *, unflatten: bool = False) → viper.collections.Task[source]

Initialize item from the given dict.

Parameters

dict (dict) – The dictionary containing the properties for this object.

Example

Host.from_dict({"ip": "1.2.3.4"})
classmethod from_func(funcpath: str) → CollectionType

Load the object from the given Python function.

Parameters

funcpath (str) – The path to a Python function that returns an instance of this class.

Example

Task.from_func("task.ping")
classmethod from_json(json: str, *args: Any, unflatten: bool = False, **kwargs: Any) → ItemType

Initialise a new object of this class from the given JSON string.

Parameters

  • json (str) – The JSON data to parse.

  • and **kwargs (*args) – These will be passed to json.laods.

Example

Hosts.from_json('[{"ip": "1.1.1.1"}]')
hash() → int

Get the hash value.

By convention all Viper native objects are frozen and thus hashable.

Return type

int

pipe(handler: Any, *args: Any) → T

Pipe this object to the given function.

Parameters

  • handler (callable) – A callable that expects this object.

  • args (object) – Arguments to be passed to the handler for decision making.

Example

Hosts.from_items(Host("1.2.3.4")).pipe(hosts2csv)
post_run = None
pre_run = None
results() → viper.collections.Results[source]

Get the past results of this task.

Return type

Rviper.collections.esults

retry = 0
stderr_processor = None
stdout_processor = None
timeout = None
to_dict(*, flatten: bool = False) → Dict[str, object][source]

Represent the item as dict.

Parameters

flatten (bool) – If true, the dict won’t be nested.

Return type

dict

Example

Host("1.2.3.4").to_dict()
to_json(*args: Any, flatten: bool = False, **kwargs: Any) → str

Represent the collection as JSON data.

Parameters

*args and **kwargs (object) – These will be passed to json.laods.

Example

Host("1.2.3.4").to_json(indent=4)
class viper.WhereConditions[source]

Bases: enum.Enum

Where query conditions for viper Items.

Example

Hosts.from_items(
    Hosts('1.1.1.1'),
    Hosts('2.2.2.2'),
).where("ip", WhereConditions.is_, '1.1.1.1')
contains = 'CONTAINS'
endswith = 'ENDSWITH'
is_ = 'IS'
is_not = 'IS_NOT'
not_contains = 'NOT_CONTAINS'
not_endswith = 'NOT_ENDSWITH'
not_startswith = 'NOT_STARTSWITH'
startswith = 'STARTSWITH'