Если занимаетесь автотестами на python, часто работаете с многопоточностью и хотите уменьшить количество boiler-plate кода – имеет смысл посмотреть на библиотеку easypy.  

В очередном тестовом фреймворке на pytest наткнулся на удобный способ работы с многопоточностью, сначала подумал, что это свой велосипед, но оказалось, что внешняя зависимость. Звезд на гитхабе мало, документации тоже почти нет, но мне понравился подход по работе с многопоточностью и еще пара занятных штук.

Сама библиотека представляет из себя набор не сильно связанных модулей, предоставляющих удобные обертки над разнообразным boiler-plate кодом. Хорошо подходит под задачи возникающие в процессе написания автотестов. Багов в библиотеке явных найти не удалось - все работает стабильно. Однако, не уверен, что использовать в production коде продукта хорошая идея – monkey-patching используется повсюду, не очень много unit тестов, мало документации, есть вероятность себе что-нибудь отстрелить.

Пройдемся по частям библиотеки, начиная с тех, которые используются в автотестах чаще всего. 

easypy.concurrency

Самый полезный модуль, из-за него я, собственно, и решил этой библиотекой поделиться, как минимум, подобный подход можно реализовать и у себя. Класс MultiObject позволяет делать вот такое:

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))

Вывод будем таким:

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

В качестве аргумента конструктору передается коллекция элементов. С помощью метода .call() указывается какой метод вызывать для каждого из этих элементов. Каждый вызов будет осуществлен в отдельном потоке. С помощью параметра workers можно указать количество потоков. По умолчанию, количество потоков равно количеству элементов.

Кроме этого, можно вызывать напрямую методы переданных объектов (объекты должны иметь одинаковый интерфейс):

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')

Вывод:

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

Метод send() является методом экземпляра класса Server. Библиотека сама вызывает этот метод для каждого экземпляра. Каждый вызов происходит в отдельном потоке, отрегулировать количество можно с помощью параметра workers. 

 Для сравнения, код на чистом threading для такой же ситуации будет выглядеть как-то так:

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]

Метод concurrent()

Следующий полезный метод, можно использовать в качестве контекстного менеджера:

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')

Вывод:

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

Можно использовать и напрямую - для запуска задачи в фоне. Если не указать параметры loop и sleep, метод внутри будет вызван один раз:

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

Оба варианта – и метод concurrent и класс MultiObject очень удобно использовать, когда тестируемая система состоит из нескольких узлов и с этими узлами нужно что-то одновременно делать. Можно также использовать для failover тестов, когда во время какого-то тестового сценария с системой что-то идет не так – например, постоянно перезагружается какой-то случайный узел в системе, а мы параллельно с этой системой работаем. 

easypy.collections

Класс ListCollection - является надстройкой над стандартным list, добавляя несколько удобных методов и поддержку класса MultiObject:

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() позволяет выбрать один элемент из коллекции, указывая либо предикат, либо фильтр (значение какого-то поля объекта). select() выбирает подмножество элементов из коллекции. Также есть ряд методов типа shuffled(), sorted(), filtered(), without() которые отвечают за простые, но удобные, операции над коллекцией.
 
Через параметр M можно создать объект MultiObject и сразу вызвать на всей коллекции метод send(), который выполнится в разных потоках. В результате получается такая конструкция:

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

Опять же - очень удобно при работе с многоузловой системой - узлы можно хранить в ListCollection и выполнять операции в разных потоках.

easypy.units

Небольшой модуль с константами для измерения времени и информации. Избавляет от необходимости писать свои, пример -

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

Поддерживаются математические операции - удобно оперировать.

easypy.resilience

Предлагает несколько декораторов, которые удобно использовать при операциях с нестабильной системой:

from easypy.resilience import resilient

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


print(get_number())

Позволяет подавить ошибку и вернуть значение из метода по умолчанию, даже в случае ошибки. Через аргументы acceptable и unacceptable можно контролировать на какие ошибки обращать внимание, а на какие - нет.

Следующий декоратор - @retrying повторяет вызов метода, можно указать либо количество повторов, либо время, в течение которого эти повторы будут совершены и время между повторами. Плюс также можно указать какие ошибки нужно пропускать, а какие нет.

from easypy.resilience import retrying 

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


false_assertion()

И retrying и resilient хорошо помогают при уже упомянутых failover тестах, когда нужно опрашивать систему в то время, пока ей нехорошо.

wait

Куда же без реализации 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 ждет пока указанный предикат вернет что-то кроме False или None. Если за таймаут какое-либо другое значение не возвращается – падает по таймауту с текстом, указанным в параметре message.

easypy.random

Два удобных метода, которые используются для генерации произвольных имен и текста:

>>> 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'

Нюансы

Установка - pip install real-easypy

Для использования нужно чтобы был заимпортирован библиотечный модуль logging – import easypy.logging, иначе будет ругаться ошибками при использовании любого метода из библиотеки. 

Еще раз ссылка на библиотеку - https://github.com/real-easypy/easypy