easypy – an unknown library for dealing with boiler-plate in python

If you are doing autotests in python, often work with multithreading and want to reduce the amount of boiler-plate code, it makes sense to look at the library easypy.

In the next test framework on pytest, I came across a convenient way to work with multithreading, at first I thought that this was my bike, but it turned out that it was an external dependency. There are few stars on the github, there is almost no documentation either, but I liked the approach to working with multithreading and a couple of other interesting things.

The library itself is a set of loosely coupled modules that provide convenient wrappers for a variety of boiler-plate code. Well suited for tasks that arise in the process of writing autotests. No obvious bugs could be found in the library – everything works stably. However, I’m not sure if it’s a good idea to use product code in production – monkey-patching is used everywhere, there are not many unit tests, there is little documentation, there is a chance to shoot something for yourself.

Let’s go through the parts of the library, starting with those that are used most often in autotests.

easypy.concurrency

The most useful module, because of it, I, in fact, decided to share this library, at least, a similar approach can be implemented on my own. Class MultiObject allows you to do this:

from time import sleep

from easypy.concurrency import MultiObject

def load_test_via_ip(ip: str):
    print(f'start load test via {ip}')
    sleep(2)
    print(f'end load test via {ip}')


ips = ['172.17.10.1', '172.17.10.2', '172.17.10.3']

MultiObject(ips).call(lambda ip: load_test_via_ip(ip))

The output will be like this:

start load test via 172.17.10.1
start load test via 172.17.10.2
start load test via 172.17.10.3
end load test via 172.17.10.1
end load test via 172.17.10.2
end load test via 172.17.10.3

A collection of elements is passed as an argument to the constructor. Using the method .call() specifies which method to call for each of these elements. Each call will be made in a separate thread. Using the parameter workers you can specify the number of threads. By default, the number of threads is equal to the number of elements.

In addition, you can directly call the methods of the passed objects (objects must have the same interface):

from time import sleep

from easypy.concurrency import concurrent

class Server:
    def __init__(self, server_name: str):
        self.server_name = server_name

    def send(self, msg: str):
        print(f'Sending "{msg}" to "{self.server_name}"')
        sleep(2)
        print(f'Sent "{msg}" to "{self.server_name}"')


servers = [Server('1'), Server('2'), Server('3')]
responses = MultiObject(servers, workers=1).send('Hello')

Conclusion:

Sending "Hello" to "1"
Sent "Hello" to "1"
Sending "Hello" to "2"
Sent "Hello" to "2"
Sending "Hello" to "3"
Sent "Hello" to "3"

Method send() is an instance method of the class Server. The library itself calls this method for each instance. Each call occurs in a separate thread, you can adjust the number using the parameter workers.

For comparison, pure threading code for the same situation would look something like this:

from threading import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(server.send, args=['Hello'])
        for server in servers
    ]

responses = [future.result() for future in futures]

Method concurrent()

The following useful method can be used as a context manager:

def _log():
    print(f'Logging smth {time.time()}')


with concurrent(_log, loop=True, sleep=1 * SECOND):
    for _ in range(3):
        sleep(2)
        print('Some work is being done')

Conclusion:

Logging smth 1688165534.384237
Logging smth 1688165535.389322
Some work is being done
Logging smth 1688165536.39416
Logging smth 1688165537.399373
Some work is being done
Logging smth 1688165538.403112
Logging smth 1688165539.4064589
Some work is being done

It can also be used directly – to run a task in the background. If you do not specify the loop and sleep parameters, the method inside will be called once:

concurrent(send_heartbeat, loop=True, sleep=1).start()

Both options – and the method concurrent and class MultiObject it is very convenient to use when the system under test consists of several nodes and you need to do something with these nodes at the same time. It can also be used for failover tests, when something goes wrong with the system during some test scenario – for example, some random node in the system is constantly rebooting, and we are working with this system in parallel.

easypy.collections

The ListCollection class is an add-on to the standard list, adding several convenient methods and support for the MultiObject class:

from easypy.collections import ListCollection

class Server:
    def __init__(self, server_name: str, server_number: int):
        self.server_name = server_name
        self.server_number = server_number

    def __str__(self):
        return f'Sever: "{self.server_name}"; number: {self.server_number}'


lc = ListCollection(
    [
       Server('nice_server_name_1', 1),
       Server('nice_server_name_2', 2),
       Server('nice_server_name_3', 3)
    ]
)
print(lc.choose())
print(lc.choose(server_number=2))
print(lc.select(lambda s: s.server_number > 1))

choose() allows you to select one element from the collection, specifying either a predicate or a filter (the value of some field of the object). select() selects a subset of elements from the collection. There are also a number of methods like shuffled(), sorted(), filtered(), without() which are responsible for simple, but convenient, operations on the collection.

Through the M parameter, you can create an object MultiObject and immediately call the method on the entire collection send(), which will be executed in different threads. The result is the following construction:

lc.select(lambda s: s.server_number > 1).M.send('Hello')

Again – very handy when working with a multi-node system – nodes can be stored in a ListCollection and perform operations on different threads.

easypy.units

A small module with constants for measuring time and information. Eliminates the need to write your own, for example –

>> from easypy.units import MiB, MB, GiB, GB, MINUTE, DAY
>> print(MiB/1000)
1048.576
>> print(MiB * 1000)
1000MiB
>> print(MiB * 1024)
GiB

Mathematical operations are supported – it is convenient to operate.

easypy.resilience

Provides several decorators that are useful when dealing with unstable systems:

from easypy.resilience import resilient

@resilient(default=0, acceptable=AssertionError)
def get_number():
    print('Going to raise AssertionError')
    raise AssertionError


print(get_number())

Allows you to suppress the error and return a value from the default method, even in the event of an error. Through arguments acceptable And unacceptable You can control which errors to pay attention to and which ones not.

The next decorator is @retrying repeats a method call, you can specify either the number of repetitions, or the time during which these repetitions will be made and the time between repetitions. Plus, you can also specify which errors to skip and which not.

from easypy.resilience import retrying 

@retrying(10 * SECOND, acceptable=AssertionError, sleep=1)
def false_assertion():
    print('Going to raise AssertionError')
    raise AssertionError


false_assertion()

AND retrying And resilient they help well with the already mentioned failover tests, when you need to interrogate the system while it is not feeling well.

wait

Where without the implementation of wait:

from time import sleep

from easypy.sync import wait

def something_went_wrong():
    sleep(1)


wait(5 * SECOND, something_went_wrong, sleep=1, message=f"Something went wrong")

wait waits for the specified predicate to return something other than False or None. If no other value is returned during the timeout, it falls by timeout with the text specified in the message parameter.

easypy.random

Two handy methods that are used to generate arbitrary names and text:

>>> from easypy.random import random_nice_name
>>> random_nice_name()
'fastidious-vulture'
>>> random_nice_name()
'balanced-skate'

>>> from easypy.random import random_string
>>> import string
>>> random_string(length=10, charset=string.ascii_lowercase)
'ejwzbrgvtd'
>>> random_string(length=10, charset=string.ascii_lowercase)
'zecalshoji'

Nuances

Installation – pip install real-easypy

To use it, you need to import the logging library module – import easypy.loggingotherwise it will swear by errors when using any method from the library.

Another link to the library – https://github.com/real-easypy/easypy