Метка: decorator

Библиотека schedule – CRON на Python

Вам приходилось работать с CRON? Это такой сервис в nix-системах, который позволяет регулярно в определенные моменты времени запускать скрипты или программы. Штука с долгой историей, в наследство которой достался странный синтаксис для описания правил:

0 * * * * my_script

Что если бы мы хотели иметь свой CRON внутри программы Python, чтобы в нужные моменты времени вызывать функции? Да еще, чтобы у него был человеческий синтаксис? Такая библиотека есть и называется schedule.

pip install schedule

Рассмотрим пример:

import schedule
import time

def job():
    print("Работаю")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

# нужно иметь свой цикл для запуска планировщика с периодом в 1 секунду:
while True:
    schedule.run_pending()
    time.sleep(1)

Как видите, правила для задания временных интервалов прекрасно читаются, словно они предложения на английском языке. Перевод пары примеров:

# спланируй.каждые(10).минут.сделать(работу)
schedule.every(10).minutes.do(job)

# спланируй.каждый().день.в(10:30).сделать(работу)
schedule.every().day.at("10:30").do(job)

В задания можно передавать параметры вот так:

def greet(name):
    print('Hello', name)

schedule.every(2).seconds.do(greet, name='Alice')

Если по какой-то причине нужно отменить задание, это делается так:

def job1():
    # возвращаем такой токен, и это задание снимается с выполнения в будущем
    return schedule.CancelJob

schedule.every().day.at('22:30').do(job1)

Если нужно отменить группу заданий, то к ним добавляют тэги:

schedule.every().day.do(greet, 'Monica').tag('daily-tasks')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks')

schedule.clear('daily-tasks')  # массовая отмена по тэгу

Метод to позволяет задать случайный интервал для выполнения задания, например от 5 до 10 секунд:

schedule.every(5).to(10).seconds.do(my_job)

Библиотека сама не обрабатывает сама исключения в ваших задачах, поэтому, возможно, понадобится создать подкласс планировщика, как в этом примере. Или декоратор, который будет отменять работу, если произошло исключение. Вот так:

import functools

# декоратор для ловли исключений
def catch_exceptions(cancel_on_failure=False):
    def catch_exceptions_decorator(job_func):
        @functools.wraps(job_func)
        def wrapper(*args, **kwargs):
            try:
                return job_func(*args, **kwargs)
            except:
                import traceback
                print(traceback.format_exc())
                if cancel_on_failure:
                    return schedule.CancelJob
        return wrapper
    return catch_exceptions_decorator

@catch_exceptions(cancel_on_failure=True)
def bad_task():
    # даст исключение, но декоратор просто отменит эту задачу
    return 1 / 0  

schedule.every(5).minutes.do(bad_task)

Если задания занимают продолжительное время или должны выполняться параллельно, то вам самостоятельно придется организовать их выполнение в отдельных потоках.

import threading
import time
import schedule

# код задания
def job():
    print("Выполняюсь в отдельном потоке")


def run_threaded(job_func):
    job_thread = threading.Thread(target=job_func)
    job_thread.start()


schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)


# бесконечный цикл, проверяющий каждую секунду, не пора ли запустить задание 
while 1:
    schedule.run_pending()
    time.sleep(1)

Celery

Если вы используете в проекте Celery, то, вероятно, вам не нужен schedule. В Celery и так есть отличный CRON:

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # эта функция выполнится при запуске - настроим вызовы задачи test

    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

    # в 7 30 по понедельникам
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

Подробнее тут.

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Абстрактный класс ABC

Просто лого для статьи

Абстрактный класс – класс, содержащий один и более абстрактных методов.

Абстрактный метод – метод, который объявлен, но не реализован.

Абстрактный класс не может быть инстанциирован (создан его экземпляр). Нужно наследовать этот класс и реализовать (переопределить) все абстрактные методы, и только после этого можно создавать экземпляры такого наследника.

В Python нет синтаксической поддержки абстрактных классов, но есть встроенный модуль abc (расшифровка – abstract base classes), который помогает проектировать абстрактные сущности.

Абстрактный класс наследуют от ABC (Python 3.4+) или указывают метакласс ABCMeta (для Python 3.0+):

from abc import ABC, ABCMeta
class Hero(ABC):
    ...

# или:
class Hero(metaclass=ABCMeta):
    ...

Любой из вариантов работает, первый современнее и короче. На данном этапе мы можем создавать объекты этих классов, потому что в них пока не абстрактных методов. Добавим:

from abc import ABC, abstractmethod
class Hero(ABC):
    @abstractmethod
    def attack(self):
        pass

Hero() – выдаст ошибку "TypeError: Can't instantiate abstract class Hero with abstract methods attack", которая говорит, что в классе Hero есть абстрактный метод attack. Мы вставили в него заглушку pass, но вообще там может быть какая-то реализация. Отнаследуем от героя Hero – конкретный подкласс лучника Archer:

class Archer(Hero):
    def attack(self):
        print('выстрел из лука')
Archer().attack()

Вот объект Archer мы можем уже создать и использовать реализацию метода attack

Кроме обычных методов, абстрактными можно обозначить и статические, классовые методы, а также свойства:

class C(ABC):
   @classmethod
   @abstractmethod
   def my_abstract_classmethod(cls):
       ...

   @staticmethod
   @abstractmethod
   def my_abstract_staticmethod():
       ...

   @property
   @abstractmethod
   def my_abstract_property(self):
       ...

   @my_abstract_property.setter
   @abstractmethod
   def my_abstract_property(self, val):
       ...

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

Формально говоря, абстрактные классы для Python не являются чем-то необходимым в силу динамичности языка. Если мы выкинем все упоминания абстрактности классов и методов из рабочего кода, он продолжит работать, как и ранее. Абстрактные классы нужны на этапе проектирования или расширения кода, чтобы обеспечивать «правильные» взаимодействия новых классов, защищая от создания экземпляров абстрактных классов. Важно помнить, что эта защита срабатывает на этапе выполнения программы, а не компиляции, как в языках Java, C++ или C#!

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Переопределение свойств класса

В заметке расскажу, как переопределить свойства (@property) в классе-наследнике. Как оказалось, это не тривиально и существуют несколько вариантов, различных между собой.

Мем про property

Допустим есть базовый класс со свойством:

class Base:
    def __init__(self):
        self._x = 100

    @property
    def x(self):
        print('Base get x')
        return self._x

    @x.setter
    def x(self, value):
        print('Base set x')
        self._x = value

Ситуация А. В классе-наследнике мы хотим переопределить ТОЛЬКО сеттер, чтобы он делал что-то еще помимо того, что умеет в базовом классе. Это не так и тривиально:

class Derived1(Base):
    @Base.x.setter
    def x(self, value):
        print('Derived1 set x')
        Base.x.fset(self, value)

Ситуация B. Хотим переопределить ТОЛЬКО геттер:

class Derived3(Base):
    @Base.x.getter
    def x(self):
        print('Derived3 get x')
        return super().x

Ситуация C. Хотим переопределить и геттер, и сеттер.  

class Derived2(Base):
    @property
    def x(self):
        print('Derived2 get x')
        return super().x
    @x.setter
    def x(self, value):
        print('Derived2 set x')
        Base.x.fset(self, value)

В этом случае мы определяем свойство как бы с нуля. Старый геттер вызывается при доступе к super().x, а старый сеттер вызываем аналогично с ситуацией A. Разница только в @x.setter вместо @Base.x.setter (по-старому не работает, проверял).

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

Как сделать проще? Писать все геттеры и сеттеры явно, а потом делать из них property; или вообще отказаться от property.

class Base:
    def __init__(self):
        self._x = 0
    def set_x(self, v):
        print('Base set x')
        self._x = v
    def get_x(self):
        print('Base get x')
        return self._x
    x = property(get_x, set_x)

class Derived(Base):
    def set_x(self, v):
        print('Derived set x')
        super().set_x(v)
    def get_x(self):
        print('Derived get x')
        return super().get_x()
    x = property(get_x, set_x)

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Полезные декораторы

Как и обещал, приведу список полезных декораторов. Среди них как стандартные поставляемые вместе с Python, так и декораторы из других библиотек и исходные коды прочих интересных декораторов.

Начнем с самых известных.

Класс-декоратор и декоратор класса

Эти две темы не так близки, как кажется, но я не мог разнести их в разные посты, лишая себя такого заголовка. Узнаем, как из класса сделать декоратор, и как написать декоратор для класса. Код примеров доступен в GIST под каждым из разделов.

Класс как декоратор

Если у класс реализовать магический метод __call__ , то экземпляр такого класса можно будет вызывать как функцию, при этом, очевидно, будет вызываться просто этот самый магический метод. Такой объект называют функтором. Пример:

class Functor:
    def __call__(self, a, b):
        print(a * b)

f = Functor()
# вызов как будто функция
f(10, 20)

Как мы помним из https://tirinox.ru/parametric-decorator/ , справа от собачки в декораторе может стоять не только функция-декоратор, но любой вызываемый объект, например, функтор. __call__, которого будет принимать на вход единственный параметр – декорируемую функцию. На примере того же декоратора-повторителя вызовов:

from functools import wraps

class Repeater:
    def __init__(self, n):
        self.n = n

    def __call__(self, f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            for _ in range(self.n):
                f(*args, **kwargs)
        return wrapper

@Repeater(3)
def foo():
    print('foo')

foo() 
# foo
# foo
# foo

Обратите внимание, что получился сразу декоратор с параметрами, где параметры – это всего лишь аргументы конструктора (нам ведь нужен экземпляр класса, а не сам класс). Поэтому если вы пишите сложный декоратор да еще и с параметрами, стоит присмотреться к его реализации классом-функтором, чтобы избежать трехэтажных вложенных функций.

Код здесь https://gist.github.com/tirinox/b6fd34de1b9de229ec2666f160c1ad82.

Декоратор для класса

Так как в Python классы создаются динамически по время интерпретации исходного кода, то можно влиять на этот процесс, например, путем декорирования. Аналогично декораторам функций, декоратор класса призван модифицировать поведение и содержание класса, не изменяя его исходный код. Похоже на наследование, но есть отличия:

  1. Декоратор класса имеет более глубокие возможности по влиянию на класс, он может удалять, добавлять, менять, переименовывать атрибуты и методы класса. Он может возвращать совершенно другой класс.
  2. Старый класс «затирается» и не может быть использован, как базовый класс при полиморфизме
  3. Декорировать можно любой класс одним и тем же универсальный декоратором, а при наследовании – мы ограничены иерархией классов и должны считаться с интерфейсами базовых классов.
  4. Презираются все принципы и ограничения ООП (из-за пунктов 1-3).

Декораторы классов полезны, чтобы внедриться в класс (иногда незаметно) и массово воздействовать на его методы и атрибуты. Типичный пример – создадим декоратор, который будет измерять время выполнения каждого метода класса. При этом сам класс никаких изменений не претерпит и не будет знать, что за ним следят:

import time

# это вспомогательный декоратор будет декорировать каждый метод класса, см. ниже
def timeit(method):
    def timed(*args, **kw):
        ts = time.time()
        result = method(*args, **kw)
        te = time.time()
        delta = (te - ts) * 1000
        print(f'{method.__name__} выполнялся {delta:2.2f} ms')
        return result
    return timed


def timeit_all_methods(cls):
    class NewCls:
        def __init__(self, *args, **kwargs):
            # проксируем полностью создание класса
            # как создали этот NewCls, также создадим и декорируемый класс
            self._obj = cls(*args, **kwargs)

        def __getattribute__(self, s):
            try:
                # папа, у меня есть атрибут s?
                x = super().__getattribute__(s)
            except AttributeError:
                # нет сынок, это не твой атрибут
                pass
            else:
                # да сынок, это твое
                return x

            # объект, значит у тебя должен быть атрибут s
            attr = self._obj.__getattribute__(s)

            # метод ли он?
            if isinstance(attr, type(self.__init__)):
                # да, обернуть его в измеритель времени
                return timeit(attr)
            else:
                # не метод, что-то другое
                return attr
    return NewCls


@time_all_class_methods
class Foo:
    def a(self):
        print("метод a начался")
        time.sleep(0.666)
        print("метод a кончился")


f = Foo()
f.a()

# метод a начался
# метод a кончился
# a 668.74 ms

Рассмотрим подробно части кода. timeit – это простой декоратор для функций, мы его уже умеем делать. Он нужен для того, чтобы декоратор класса timeit_all_methods обернул в timeit каждый метод декорируемого класса.

Декоратор timeit_all_methods содержит в себе определение нового класса NewCls и возвращает его вместо оригинального класса. Т.е. класс Foo – это уже не Foo, а NewCls. Конструктор класса NewCls принимает произвольные аргументы (ведь нам не известно заранее, какой конструктор у Foo, и у любого другого класса, который мы декорируем). Поэтому конструктор просто создает поле, где будет хранить экземпляр оригинального класса, и передает ему в конструктор все свои аргументы.

Самый сложный метод – __getattribute__ – он полон магии. Он вызывается, когда кто-то пытается обратиться как какому угодно атрибуту (полю, методы и т. п.) класса NewCls. Первым делом мы должны обратиться к своему родителю super() и спросить у него, не обладаем ли мы сами атрибутом, который проверяем. Именно к родителю, чтобы избежать рекурсии (иначе мы попадем в тот же метод, в котором уже находимся)! Если это наш атрибут (атрибут класса декоратора) – вернем его сразу, с ним ничего не надо делать. Иначе, вероятно, это атрибут исходного класса – получим его у него. И проверим его тип, сравним его с типом любого метода. Если тип – метод (bound method), то обернем его в декоратор timeit и вернем, иначе (это не метод, а свойство или статический метод) – вернем без изменений.

Таким образом мы проксируем все атрибуты обернутого класса через NewCls, оборачивая в timeit только методы.

Задание на дом: создать класс декоратор класса, иначе говоря скрестить два раздела статьи и сделать класс-функтор, который может декорировать другой класс. Идея: декоратор, который измеряет время выполнения каждого метода, и печатает предупреждение, только если время выполнения было больше критического (параметр):

@TimeItCritical(critical_time=0.3)
class Foo:
    def a(self):
        print("медленный метод начался")
        time.sleep(1.0)
        print("медленный метод кончился")

    def b(self):
        time.sleep(0.1)
        print('быстрый метод')

f = Foo()
f.a()
f.b()

# медленный метод начался
# медленный метод кончился
# a выполнялся медленно 1.0011 s
# быстрый метод

Код доступен в https://gist.github.com/tirinox/507258b36e77dfec1448f8cf1d259356

🤩 Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway! 👈