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()
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()
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)
# 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)
# Проверяем, пингуется ли машина
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()
import uuid
class Fruit:
def __init__(self, name, id=uuid.uuid4(), color='Green'):
self.name = name
self.id = id
self.color = color
import uuid
from dataclasses import dataclass
@dataclass
class Fruit:
name: str
id: int = uuid.uuid4()
color: str = 'Green'
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'
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
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)
# 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)
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!")
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}')
# 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)
# 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)
# 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)
bool_items:
false_item: false
true_item: true
int_items:
- 1
- 2
- 3
- 4
- 5
str_item: Im a string
# 2. Read *.yaml file content to dict
with open('my_yaml_dump.yml', 'r') as f:
loaded_from_file_dict = yaml.safe_load(f)
(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]
ignore=
# Line too long
E501,
#Line break after binary operator
W504
# Could also be ommited via commandline param: --exclude ./venv
exclude = ./venv
# 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