Подписаться
Главная / Блог / PYTHON code samples

Полезные кусочки
кода на Python

Python code samples /
Полезные кусочки
кода на Python

Smartiqa Article Python code samples
  • Дата: 16 августа 2023
  • Автор: Евгений Поваров
  • Git репозиторий: useful_samples
1

1. Retry method / Функция, реализующая повторные попытки выполнения кода

Для чего нужно?
Иногда часть кода может падать с ошибкой, и эта ситуация является ожидаемой для разработчика. В таком случае может быть необходимо повторить выполнение упавшего кода. Наиболее типичный пример - отправка HTTP запроса на сервер. Сервер может быть недоступен всего на долю секунды - но именно в этот момент мы можем попытаться отправить запрос, и как результат, он завершится с ошибкой. В таких случаях повторная отправка того же запроса является оптимальным решением. Ведь в данном случае то, что наш код упал, не зависит от него самого, а зависит от внешних факторов, которые мы не можем изменить.

При реализации retry метода необходимо ответить на следующие вопросы:
  1. Сколько попыток необходимо совершить? 2, 5 или 20?
  2. Хотим ли мы делать временные паузы между попытками? Если да, то какова длина этих пауз? Длина представляет собой неизменное значение (например, между всеми попытками мы ждем 5 секунд) или ее величина растет с увеличением числа попыток? То есть например, мы можем сначала делать попытки чаще, но потом все реже, чтобы не выполнять лишнюю работу и не закидывать бесполезными запросами еле живой сервер).
  3. Что является условием завершения наших попыток? Самое первое и желаемое - попытка проходит успешно. Но что, если код продолжает падать, а положительного результата все нет? Необходимо решить, когда остановить попытки. Здесь два основных варианта: 1) ограничить их количество, 2) ограничить время - то есть задать timeout, после которого мы останавливаем ретраи.
  4. Что делать, если все попытки оказались неуспешными? В большинстве случаев мы захотим упасть с ошибкой, но возможны и варианты, когда проблему можно проигнорировать и ограничиться выводом варнинга в консоль нашего приложения.
Retry method (Python)

import time

# Максимальное количество попыток
RETRIES = 3
# Промежуток времени (60 сек), в течение которого мы будем ретраиться
TIMEOUT = 60
# Частота, c которой будем ретраиться
PERIOD = 5

# Функция-декоратор, которая модифицирует переданный ей метод
# В нашем случае модификация заключается в повторении вызовов переданного метода
def retry(max_retries, timeout, period):
    def outer(func):
        def inner(*args, **kwargs):
            # Задаем время, когда необходимо остановить попытки
            end_time = time.time() + timeout
            # Создаем счетчик, с каждой попыткой будем уменьшать его значение на 1
            retries = max_retries
            # Бесконечно крутимся в цикле до тех пор, пока не случится одно из событий: 
            # 1) метод func() выполнится успешно
            # 2) закончится отведенное на ретраи время
            # 3) исчерпаются попытки
            while True:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f'{e}')
                    if time.time() > end_time:
                        raise 'Timeout has expired!'
                    if retries == 1:
                        raise e
                    else:
                        retries -= 1
                        print(f"Attempts left: {retries}")
                        print(f"Sleeping {period} seconds ..." )
                        time.sleep(period)
        return inner
    return outer

# Протестируем наш декоратор на примере падающей функции
@retry(RETRIES, TIMEOUT, PERIOD)
def send_request():
    raise Exception('Code block has failed. This is expected.')

# Тесты
send_request()
Smartiqa retry method
И еще рассмотрим следующую модификацию нашего retry() метода - будем ретраить не все ошибки подряд (сейчас мы делаем повторные попытки для всех Exception), а выборочно. В частности будем ретраиться, если:
  1. Было падение c эксепшеном, переданным через параметр exception.
  2. Было падение с любым типом эксепшена, но в тексте ошибки содержится нужная фраза (передается через параметр error_msg).
Modified retry (Python)

def modified_retry(max_retries, timeout, exception=None, error_msg=None, period=1):
    def outer(func):
        def inner(*args, **kwargs):
            end_time = time.time() + timeout
            retries = max_retries
            while retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if (e and e == exception) or (error_msg and error_msg in str(e)):
                        log.error(f'ERROR: {e}')
                        if time.time() > end_time:
                            log.error(f'Failed to successfully execute cmd during {timeout} sec')
                            raise e
                        if retries == 1:
                            log.error(f'Failed to successfully execute cmd with {max_retries} attempts')
                            raise e
                        retries -= 1
                        time.sleep(period)
                    else:
                        raise e
        return inner
    return outer

# Будем делать повторные попытки, только если упали с TimeoutExpired
@retry(5, 60, exception=subprocess.TimeoutExpired)
def send_request():
    raise TimeoutExpired('Operation timed out. This is expected.')

# Тест
send_request()
Смотреть код на Github
2

2. Wait method / Функция, ожидающая успешное выполнение условия

Для чего нужно?
Иногда в ходе работы программы нужно дождаться какого-то события, чтобы продолжить выполнение кода. Иногда количество времени, которое нужно подождать заранее, известно. А иногда - нет. И так бывает чаще всего. Рассмотрим следующий пример. Предположим, что программа ждет поднятия HTTP сервера, чтобы она могла отправлять ему запросы и получать ответы. В зависимости от разных внешних факторов время поднятия сервера может варьироваться от 2 минут до 15. Какие есть варианты решения?

  1. Можем подождать 15 минут. И сразу после их окончания начать слать запросы. Этот вариант довольно простой в реализации, но у него есть недостатки. В частности, в случае если сервер поднимется через 3 минуты, основная программа будет терять время впустую, выжидая положенные 15 минут. Так же возможна ситуация, когда серверу понадобится не 15 минут для загрузки, а 17. Тогда мы начнем слать запросы раньше времени - еще до поднятия сервера.
  2. Можем реализовать механизм ожидания. В течение определенного времени с заданной периодичностью будем пытаться слать запросы на сервер. Если запрос неуспешный - основная программа считает этот Exception ожидаемым и повторит попытку позднее. Это будет продолжаться до тех пор, пока не выполнится одно из условий: 1) запрос вернет код 200 (будет успешным) или 2) мы достигнем таймаута.

Вопросы, на которые нужно ответить при создании wait_until() метода:
  1. С какой периодичностью будем проверять, что мы дождались необходимого условия? Иногда период, с которым мы выполняем проверку, может быть постоянной величиной (например, можем опрашивать сервер каждые 10 секунд). А может быть величиной изменяемой. Например, она может расти в геометрической прогрессии.
  2. В какой момент мы решаем, что пора сдаваться? Кажется, что чем больше таймаут, тем лучше - шансы дождаться больше. Однако в некоторых случаях слишком долгий таймаут может напрасно тратить время работы программы - если бы мы раньше упали с ошибкой, то и раньше смогли бы что-нибудь предпринять для решения проблемы.

Wait until method (Python)

import time

# Главная функция
def wait_until(condition, description, timeout=300, period=0.25, *args, **kwargs):
    final_time = time.time() + timeout
    while time.time() < final_time:
        # Вызываем переданную функцию
        # Если ее возвращаемое значение эквивалентно True, то ожидание завершается
        if condition(*args, **kwargs):
            return True
        # Если нет - ждем дальше
        time.sleep(period)
    raise TimeoutError(f'Timed out waiting for condition: [{description}]')


# Через замыкание создаем метод, который отсчитывает n секунд
def closure(n):

    def sleep_several_seconds():
        nonlocal n
        time.sleep(1)
        n = n - 1
        print(f"{n} seconds left")
        return n

    return sleep_several_seconds


# Метод, который проверяет, что отведенное количество секунд закончилось
def five_seconds_passed():
    return sleep_five_seconds() == 0


# Создаем метод, который будет отсчитывать 5 секунд
sleep_five_seconds = closure(5)
Tests

# 1. Позитивный кейс
# Ждем, пока будут отсчитаны 5 секунд с таймаутом 10 секунд - не выходим за пределы таймаута
wait_until(condition=five_seconds_passed, description='Five seconds are over', timeout=10)

# 2. Негативный кейс
# Ждем, пока будут отсчитаны 5 секунд с таймаутом 3 секунды - превышаем таймаут и падаем с TimeoutError
wait_until(condition=five_seconds_passed, description='Five seconds are over', timeout=3)
Smartiqa wait_until method
Smartiqa wait_until method
Еще один пример использования - ждем, пока машина станет доступна в сети (пингуем по IP адресу):
Wait until machine is up (Python)

# Проверяем, пингуется ли машина
def host_is_pingable(ip):
    return os.system(f"ping -c 1 {ip}")

def wait_until(condition, description, timeout=300, period=5, *args, **kwargs):
    final_time = time.time() + timeout
    while time.time() < final_time:
        output = condition(*args, **kwargs)
        if output:
            return output
        time.sleep(period)
    raise TimeoutError(f'Timed out waiting for condition: [{description}]')

# Ждем, пока машина станет доступна в сети
def wait_until_host_is_pingable(ip):
    wait_until(condition=host_is_pingable, ip=ip, timeout=60, period=5, description=f"Host {ip} is up")

# Тест
wait_until_host_is_pingable()
Смотреть код на Github
3

3. Data classes / Классы по работе с данными

Data classes в Python позволяют упростить-укоротить написание классов, которые работают с данными. Эта возможность появилась в Python 3.7. Давайте напишем класс Fruit, который будет хранить информацию о фруктах:
Class Fruit (Python)

import uuid

class Fruit:

    def __init__(self, name, id=uuid.uuid4(), color='Green'):
        self.name = name
        self.id = id
        self.color = color
А теперь напишем то же самое, но с применением декоратора @dataclass:
Data class Fruit (Python)

import uuid
from dataclasses import dataclass

@dataclass
class Fruit:
    name: str
    id: int = uuid.uuid4()
    color: str = 'Green'
Обратите внимание, что для некоторых полей заданы значения по умолчанию. Дефолтным значением может быть изменяемая величина, и даже можно задать список допустимых значений. Например, вот так:
Data class Fruit (Python)

import uuid
from dataclasses import dataclass
from typing import Literal

@dataclass
class Fruit:
    name: str
    id: int = uuid.uuid4()
    color: Literal['Green', 'Red', 'Orange', 'Blue', 'Yellow', 'Black', 'White'] = 'Green'
И далее приводим код полностью плюс тест:
Data class Fruit (Python)

from dataclasses import dataclass
from typing import Literal
import uuid


# class Fruit:
#
#     def __init__(self, name, id=uuid.uuid4(), color='Green'):
#         self.name = name
#         self.id = id
#         self.color = color

@dataclass
class Fruit:
    name: str
    id: int = uuid.uuid4()
    color: Literal['Green', 'Red', 'Orange', 'Blue', 'Yellow', 'Black', 'White'] = 'Green'


# Тест
fruits = [
    Fruit(name='Peach', id=1, color='Orange'),
    Fruit(name='Orange', id=2, color='Orange'),
    Fruit(name='Banana', color='Yellow'),
    Fruit(name='Apple')
]

for fruit in fruits:
    print(fruit)

# Результат
# /Users/tati/Documents/Projects/inherit_composition/venv/bin/python 
# /Users/tati/Documents/Projects/Pycharm/demo/dataclass.py
# Fruit(name='Peach', id=1, color='Orange')
# Fruit(name='Orange', id=2, color='Orange')
# Fruit(name='Banana', id=UUID('b4022f39-6a61-463b-a7cb-71a744f5b4df'), color='Yellow')
# Fruit(name='Apple', id=UUID('b4022f39-6a61-463b-a7cb-71a744f5b4df'), color='Green')
# Process finished with exit code 0
Smartiqa data class
Смотреть код на Github

Читайте также

[ Часть 2 ] Собеседование на должность QA Automation Engineer. Основы программирования.
Вопросы и ответы по темам: проектирование, языки программирования, ООП, алгоритмы, задачи, Python
4

4. *args **kwargs / Переменное количество аргументов функции

Для справки:
  1. *args - список НЕименованных параметров. Пример: [1, 2, 3]
  2. **kwargs - словарь именованных параметров. Пример: {'a': 1, 'b': 2, 'c': 3]
Простой пример использования таких параметров:
Example 1 (Python)

def my_func_args(*args):
    # [1, 2, 3]
    pass

def my_func_kwargs(**kwargs):
    # {'a': 1, 'b': 2, 'c': 3}
    pass

my_func_args(1, 2, 3)
my_func_kwargs(a=1, b=2, c=3)
Example 2. Count items in bag (Python)

# Counts arguments quantity
def count_items_in_bag(*args):
    sum = 0
    for item in args:
        print(f'Counting {item}')
        sum += 1
    print(f'Total items: {sum}')
    return

def count_named_items_in_fridge(**kwargs):
    sum = 0
    for name, quantity in kwargs.items():
        print(f'Counting {name}={quantity}')
        sum += quantity
    print(f'Total items: {sum}')
    return

count_items_in_bag('phone', 'credit card', 'pen', 'notebook', 'handkerchief')
count_named_items_in_fridge(milk=1, apple=5, eggs=10, butter=1)
args kwargs example
А теперь давайте попробуем передавать именованные и позиционные параметры в разном порядке:
Example 3. Use positional params in different order (Python)

def print_args(a, b=0, **kwargs):
    print("Lets print args:")
    print(f'a={a}')
    print(f'b={b}')
    for key, value in kwargs.items():
        print(f'{key}={value}')

# The first 2 args could be unnamed since they are passed in the initial order
print_args(1, 2, named_arg=100, one_more_named_arg="I'm one more named arg!")
# We can change args order - but we need to specify their names in such case
print_args(named_arg=100, b=2, a=1, one_more_named_arg="I'm one more named arg!")
# We can skip 'b' param value since it already has default one (b=0)
print_args(named_arg=100, a=1, one_more_named_arg="I'm one more named arg!")
args kwargs example
И еще давайте создадим функцию, которая будет принимать другую функцию и ее параметры. Обратите внимание, что *args тут как раз кстати - так как мы заранее не знаем, какую функцию передадут, то и количество ее параметров тоже заранее неизвестно.
Example 4. Pass function and its args to another function (Python)

def print_func_name_and_call(times, func, *args):
    for iteration in range(0, times):
        print(f'Function name is "{func.__name__}" ({iteration})')
    return func(*args)


def sum(*args):
    sum = 0
    for item in args:
        sum += item
    return sum

# Here we pass sum() function name and its arguments 10, 100, 1000
s = print_func_name_and_call(3, sum, 10, 100, 1000)
print(f'Sum is {s}')
args kwargs example
Смотреть код на Github
5

5. Работа с json

Для справки:
Работаем с помощью библиотеки json. Манипуляции делим на два вида:
  1. Дампинг словаря (dict) в json-строку или файл: dict -> str / file
  2. Парсинг json из строки/файла в словарь (dict): str / file -> dict
1. Dump dict to json str or file (Python)

# 1. Dump dict to json str or file
my_dict = {
    'str_item': 'Im a string',
    'int_item': 100,
    'bool_item': True
}

# dict -> json str
my_str = json.dumps(my_dict)
my_str_with_indent = json.dumps(my_dict, indent=2)

print(my_str)
print(my_str_with_indent)

# dict -> json file
with open('my_json_dump.json', 'w') as f:
    json.dump(my_dict, f, indent=2)
json example
2. Load json from string/file to dict (Python)

# 2. Load json from string/file to dict
dict_loaded_from_str = json.loads(my_str)

with open('my_json_dump.json', 'r') as f:
    dict_loaded_from_file = json.load(f)
json example
Смотреть код на Github
6

5. Работа с yaml

Работаем с помощью библиотеки yaml. Манипуляции снова делим на два вида:
  1. Дампинг словаря (dict) в файл: dict -> yaml file
  2. Парсинг yaml из файла в словарь (dict): yaml file -> dict
1. Dump dict to yaml file (Python)

# 1. Dump dict to *.yaml file
my_dict = {
    'str_item': 'Im a string',
    'int_items': [1, 2, 3, 4, 5],
    'bool_items': {'true_item': True, 'false_item': False}
}

with open('my_yaml_dump.yml', 'w') as f:
    yaml.dump(my_dict, f)
my_yaml_dump.yml

bool_items:
  false_item: false
  true_item: true
int_items:
- 1
- 2
- 3
- 4
- 5
str_item: Im a string
2. Load from yaml file to dict (Python)

# 2. Read *.yaml file content to dict
with open('my_yaml_dump.yml', 'r') as f:
    loaded_from_file_dict = yaml.safe_load(f)
yaml example
Смотреть код на Github
7

7. Syntax and code style check / Проверка кода на наличие синтаксических и стилевых ошибок

Проверять код будем с помощью библиотеки flake8. Установка через консоль (Linux):
Install flake8 (Linux terminal)

(venv) tati@tati useful_samples % pip3 install flake8
Collecting flake8
Successfully installed flake8-6.0.0 mccabe-0.7.0 pycodestyle-2.10.0 pyflakes-3.0.1
Если запускать flake8 без параметров и настроек, то получим довольно много лишних ошибок в стиле 'E501 line too long'. Поэтому перед первым запуском необходимо сделать две вещи:
  1. Создать конфигурационный файл (например, .flake8 в корневой директории проекта) и указать в нем, какие ошибки/варнинги мы хотим игнорировать. Указать путь к файлу при запуске (если он не подцепляется автоматически)
  2. В конфиге или при запуске учесть, в каких директориях/файлах мы не хотим проверять ошибки (например, не стоит проверять сторонние библиотеки из папки venv).
Config file .flake

[flake8]
ignore=
    # Line too long
    E501,
    #Line break after binary operator
    W504
# Could also be ommited via commandline param: --exclude ./venv 
exclude = ./venv
Execution example (Linux terminal)

# Execution check before style fixes
(venv) tati@tati useful_samples % flake8 . --config .flake8 --count
./args_kwargs.py:75:21: W292 no newline at end of file
./oop_pizza.py:150:15: F541 f-string is missing placeholders
./oop_pizza.py:153:22: E225 missing whitespace around operator
./oop_pizza.py:209:22: W292 no newline at end of file
./retry.py:7:1: E302 expected 2 blank lines, found 1
./retry.py:24:63: E202 whitespace before ')'
./retry.py:43:1: W391 blank line at end of file
7

# Execution result after all errors/warnings were fixed
(venv) tati@tati useful_samples % flake8 . --config .flake8 --exclude ./venv --count
0
Как видим, в нашем случае в проекте было найдено 7 ошибок. При запуске ошибки/варнинги из папки venv были исключены. Далее все замечания были поправлены.
16 АВГУСТА / 2023
Как вам материал?
ПОКАЗАТЬ КОММЕНТАРИИ

Читайте также