O mică poveste despre decoratori

Ionel Cristian Mărieș — Partizan Python / OSS

blog.ionelmc.ro

github.com/ionelmc

Cine nu a scris un decorator?

>>> from functools import wraps
>>> def log_errors(func):
...     @wraps(func)
...     def log_errors_wrapper(*args, **kwargs):
...         try:
...             return func(*args, **kwargs)
...         except Exception as exc:
...             print("Raised %r for %s/%s" % (exc, args, kwargs))
...             raise
...     return log_errors_wrapper

>>> @log_errors
... def broken_function():
...     raise RuntimeError()

>>> from pytest import raises
>>> raises(RuntimeError, broken_function)
Raised RuntimeError() for ()/{}
...

def log_errors(func):
    def log_errors_wrapper(arg):
        return func(arg)
    return log_errors_wrapper

@log_errors
def broken_function():
    pass

broken_function = log_errors(broken_function)

>>> class log_errors(object):
...     def __init__(self, func):
...         self.func = func
...     def __call__(self, *args, **kwargs):
...         try:
...             return self.func(*args, **kwargs)
...         except Exception as exc:
...             print("Raised %r for %s/%s" % (exc, args, kwargs))
...             raise

>>> @log_errors
... def broken_function():
...     raise RuntimeError()

>>> from pytest import raises
>>> raises(RuntimeError, broken_function)
Raised RuntimeError() for ()/{}
...

A fost odată ca niciodată

Că de n-ar fi bad practice

Nu s-ar povesti

>>> def log_errors(func):
...     def wrapper(*args, **kwargs):
...         try:
...             return func(*args, **kwargs)
...         except Exception as exc:
...             print("Raised %r for %s/%s" % (exc, args, kwargs))
...             raise exc
...     return wrapper

• Ce nu e bine?

>>> @log_errors
... def foobar():
...     pass

>>> print(foobar)
<function ...wrapper at 0x...>

• Numele funcției, dat dispărut ...

>>> @log_errors
... def foobar():
...     unu()
>>> def unu():
...     doi()
>>> def doi():
...     raise Exception("Dezastru ...")

>>> foobar()
Traceback (most recent call last):
...
  File "<doctest decoratori.rst[...]>", line ..., in wrapper
...
Exception: Dezastru ...

• Care wrapper? Numele e prea generic.
• Traceback-ul nu contine informatii despre doi si unu (în Python 2)

• Mult cod repetitiv

Există 2 tipuri de funcții, decise la compilare:

• Funcția cea de toate zilele ...
• Funcția generator, dracul împielițat:
  • are yield
  • poate avea return (gol)
  • poate avea return valoare (doar în Python 3)

Așadar, funcția generator intoarce un generator.

• Dacă excepția este aruncată după ce a început iterarea atunci decoratorul nostru nu o poate prinde.
• Trebuie sa consumam generatorul (for i in ...: yield i)

>>> @log_errors
... def broken_generator():
...     yield 1
...     raise RuntimeError()

>>> raises(RuntimeError, list, broken_generator()).value
RuntimeError()

Dooh ! Decoratorul nu face nimic ...

• Otrava prescrisă: condiții și repetiții

>>> from inspect import isgeneratorfunction
>>> def log_errors(func):
...     if isgeneratorfunction(func):
...         @wraps(func)
...         def log_errors_wrapper(*args, **kwargs):
...             try:
...                 for item in func(*args, **kwargs):
...                     yield item
...             except Exception as exc:
...                 print("Raised %r for %s/%s" % (exc, args, kwargs))
...                 raise
...     else:
...         @wraps(func)
...         def log_errors_wrapper(*args, **kwargs):
...             try:
...                 return func(*args, **kwargs)
...             except Exception as exc:
...                 print("Raised %r for %s/%s" % (exc, args, kwargs))
...                 raise
...     return log_errors_wrapper

>>> @log_errors
... def broken_generator():
...     yield 1
...     raise RuntimeError()

>>> raises(RuntimeError, list, broken_generator())
Raised RuntimeError() for ()/{}
...

• Trebuie 2 functii - fiindcă funcția generator (are yield) nu poate avea return cu valoare
• Nu merge cu corutine ...

Python 3:

>>> from inspect import isgeneratorfunction
>>> def log_errors(func):
...     if isgeneratorfunction(func):
...         @wraps(func)
...         def log_errors_wrapper(*args, **kwargs):
...             try:
...                 yield from func(*args, **kwargs)
...             except Exception as exc:
...                 print("Raised %r for %s/%s" % (exc, args, kwargs))
...                 raise
...     else:
...         @wraps(func)
...         def log_errors_wrapper(*args, **kwargs):
...             try:
...                 return func(*args, **kwargs)
...             except Exception as exc:
...                 print("Raised %r for %s/%s" % (exc, args, kwargs))
...                 raise
...     return log_errors_wrapper

>>> @log_errors
... def broken_coroutine():
...     print((yield 1))
...     raise RuntimeError()

>>> coro = broken_coroutine()
>>> next(coro)
1
>>> raises(RuntimeError, coro.send, 'mesaj')
mesaj
Raised RuntimeError() for ()/{}
...

yield from (PEP-380) în Python 2? O minune (1/2):

_i = iter(EXPR) # EXPR ar fi `func(*args, **kwargs)`
try:
    _y = next(_i)
except StopIteration as _e:
    _r = _e.value
else:
    while 1:
        try:
            _s = yield _y
        except GeneratorExit as _e:
            try:
                _m = _i.close
            except AttributeError:
                pass
            else:
                _m()
            raise _e
        except BaseException as _e:
            _x = sys.exc_info()

yield from (PEP-380) în Python 2? O minune (2/2):

            try:
                _m = _i.throw
            except AttributeError:
                raise _e
            else:
                try:
                    _y = _m(*_x)
                except StopIteration as _e:
                    _r = _e.value
                    break
        else:
            try:
                if _s is None:
                    _y = next(_i)
                else:
                    _y = _i.send(_s)
            except StopIteration as _e:
                _r = _e.value
                break
RESULT = _r

>>> from aspectlib import Aspect
>>> @Aspect
... def log_errors(*args, **kwargs):
...     try:
...         yield
...     except Exception as exc:
...         print("Raised %r for %s/%s" % (exc, args, kwargs))
...         raise

>>> @log_errors
... def broken_function():
...     raise RuntimeError()
>>> raises(RuntimeError, broken_function)
Raised RuntimeError() for ()/{}
...

Mai multe detalii: documentație aspectlib.

Merge corect cu generatori:

>>> @log_errors
... def broken_generator():
...     yield 1
...     raise RuntimeError()
>>> raises(RuntimeError, lambda: list(broken_generator()))
Raised RuntimeError() for ()/{}
...

Și corutine:

>>> @log_errors
... def broken_coroutine():
...     print((yield 1))
...     raise RuntimeError()

>>> coro = broken_coroutine()
>>> next(coro)
1
>>> raises(RuntimeError, coro.send, 'mesaj')
mesaj
Raised RuntimeError() for ()/{}
...

>>> def trebuie_mecanic(func):
...     @wraps(func)
...     def wrapper_trebuie_mecanic(sofer):
...         if not sofer.are_bujie_de_rezerva:
...             raise RuntimeError("N-ai noroc")
...         return func(sofer)
...     return wrapper_trebuie_mecanic

>>> class Dacie(object):
...     @trebuie_mecanic
...     def porneste(self, sofer):
...         print("Blană !")

>>> from collections import namedtuple
>>> Sofer = namedtuple("Sofer", ["are_bujie_de_rezerva"])

>>> rabla = Dacie()
>>> rabla.porneste(Sofer(True))
Traceback (most recent call last):
...
TypeError: wrapper_trebuie_mecanic() takes 1 positional argument but 2 were given

Opaaaaaa ....

>>> class Metoda(object):
...     def __init__(self, func, nevasta):
...         self.func = func
...         self.nevasta = nevasta
...     def __call__(self, *args, **kwargs):
...         return self.func(self.nevasta, *args, **kwargs)
...     def __repr__(self):
...         return "<metodă însurată %s cu %s>" % (
...             self.func.__name__, self.nevasta)

>>> class Functie(object):
...     factory = Metoda
...     def __init__(self, func):
...         self.func = func
...     def __call__(self, *args, **kwargs):
...         return self.func(*args, **kwargs)
...     def __repr__(self):
...         return "<funcție %s>" % (self.func.__name__)
...     def __get__(self, instanta, clasa):
...         if instanta is None:
...             return self
...         return self.factory(self.func, instanta)

>>> def haleste(cine):
...     print(cine, "mânâncă ...")

Nelegată:

>>> Functie(haleste)
<funcție haleste>

>>> class Gheorghe(object):
...     manca = Functie(haleste)
>>> Gheorghe.manca # nu e 100% corect, ar trebui sa fie "unbound function ..."
<funcție haleste>

Legată:

>>> gheo = Gheorghe()
>>> gheo.manca
<metodă însurată haleste cu <__main__.Gheorghe object at ...>>
>>> gheo.manca()
<__main__.Gheorghe object at ...> mânâncă ...

>>> class MixinTrebuieMecanic(object):
...     def __call__(self, sofer):
...         if not sofer.are_bujie_de_rezerva:
...             raise RuntimeError("N-ai noroc")
...         return super(MixinTrebuieMecanic, self)(sofer)

>>> class MetodaTrebuieMecanic(Metoda, MixinTrebuieMecanic):
...     pass

>>> class TrebuieMecanic(Functie, MixinTrebuieMecanic):
...     factory = MetodaTrebuieMecanic

>>> class Dacie(object):
...     @TrebuieMecanic
...     def porneste(self, sofer):
...         print("Blană !")

>>> rabla = Dacie()
>>> rabla.porneste(Sofer(True))
Blană !

• Aceasta nu e soluția perfectă desigur, există altele ...

Fară prea mare bataie de cap:

>>> import wrapt

>>> @wrapt.decorator
... def trebuie_mecanic(func, instanta, args, kwargs):
...     sofer, = args
...     if not sofer.are_bujie_de_rezerva:
...         raise RuntimeError("N-ai noroc")
...     return func(sofer)

>>> class Dacie(object):
...     @trebuie_mecanic
...     def porneste(self, sofer):
...         print("Blană !")

>>> rabla = Dacie()
>>> rabla.porneste(Sofer(True))
Blană !

Acoperă toate cazurile. Documentație.

Prezentarea: http://bit.ly/decoratori