Application Scenarios for Decorators
- additional functionality
- Data cleansing or addition.
- Function parameter type validation @require_ints Similar pre-request interception
- Data format conversion Change function return dictionary to JSON/YAML similar to response post tampering
- Providing additional data for functions
- function registration
- Register a task in the Task Center
- Registering a function with a signal handler
Decorator implementation for different application scenarios
function registry
simple registry
funcs = [] def register(func): (func) return func @register def a(): return 3 @register def b(): return 5 # Access results result = [func() for func in funcs]
Registry isolation (using different instances of a class)
class Registry(object):
def __init__(self):
self._funcs = []
def register(self, func):
self._funcs.append(func)
def run_all(self):
return [func() for func in self._funcs]
r1 = Registry()
r2 = Registry()
@
def a():
return 3
@
def b():
return 5
@
@
Execution-time encapsulation code
Type checking
from functools import wraps
def require_ints(func):
@wraps(func) # Copy the info from func to inner
def inner(*args, **kwargs):
for arg list(args) + list(()):
if not isinstance(arg, int:
raise TypeError("{} Accept onlyinttype parameter".format(func.__name__)
return func(*args, **kwargs)
return inner
user authentication
from functools import wraps
class User(object):
def __init__(self, username, email):
= username
= email
class AnonymousUser(object):
def __init__(self):
= = None
def __nonzero__(self): # Called when converting an object to a bool type
return False
def requires_user(func):
@wraps(func)
def inner(user, *args, **kwargs): # This decorator does not support decorated classes because self is not supported in the first argument.
if user and isinstance(user, User):
return func(use, *args, **kwargs)
else:
raise ValueError("Non-legal users")
return inner
Output Formatting
import json
from functools import wraps
def json_output(func): # Convert the dictionary format returned by func to a json string.
@wrap(func)
def inner(*args, **kwargs):
return (func(*args, **kwargs))
return inner
exception capture
import json
from functools import wraps
class Error1(Exception):
def __init__(self, msg):
= msg
def __str__(self):
return
def json_output(func):
@wrap(func)
def inner(*args, **kwargs):
try:
result = func(*args, **kwargs)
except Error1 as ex:
result = {"status": "error", "msg": str(ex)}
return (result)
return inner
# Usage
@json_ouput
def error():
raise Error1("The exception is caught and output in JSON format.")
Log Management
import time
import logging
from functools import wraps
def logged(func):
@wraps(func)
def inner(*args, **kwargs): # *args can decorate functions as well as classes
start = ()
result = func(*args, **kwargs)
exec_time = () - start
logger = ("")
("{} invocation time:{:.2} execution time:{:.2}s in the end:{}".format(func.__name__, start, exec_time, result)
Decorator with parameters
Decorator with parameters is equivalent to a function that returns a decorator, @deco(a=1) will first execute deco(a=1) to get an actual decorator before calling @, Decorator with parameters deco(a=1) is executed immediately upon module import.
decorative category
Add sortable functionality to classes (without extending the parent's methods by inheriting from the subclasses, e.g., if multiple classes need to add this functionality).
import time
from functools import wraps
def sortable_by_created(cls):
original_init = cls.__init__
@wrap(original_init)
def new_init(self, *args, **kwargs):
original_init(*args, **kwargs)
self._created = ()
cls.__init__ = new_init
cls.__lt__ = lambda self, other: self._created < other._created
cls.__gt__ = lambda self, other: self._created > other._created
return cls
A SortableByCreated() class can also be defined, with children using multiple inheritance from their parent and SortableByCreated.
type conversion
Functions may become instances of a class after being decorated, in which case the __call__ method should be provided for the returned class in order to be compatible with the function call.
class Task(object):
def __call__(self, *args, **kwargs):
return (*args, **kwargs)
def run(self, *args, **kwargs):
raise NotImplementedError("The subclass does not implement the interface.")
def task(func):
class SubTask(Task):
def run(self, *args, **kwargs):
func(*args, **kwargs)
return SubTask()
Chapter 2 Context Manager
define
Packaging arbitrary code
Ensuring consistency in implementation
grammatical
with statement
__enter__ and __exit__ methods
class ContextManager(object):
def __init__(self):
= False
def __enter__(self):
= True
return self
def __exit__(self, exc_type, exc_instance, traceback):
= False
application scenario
Resource cleanup
import pymysql
class DBConnection(object):
def __init__(self, *args, **kwargs):
, = args, kwargs
def __enter__(self):
= (*args, **kwargs)
return ()
def __exit__(self, exc_type, exc_instance, trackback):
()
Exception handling (avoid duplication)
Propagate exception (return False in __exit__)
Terminate exception (return True in __exit__)
class BubleExceptions(object):
def __enter__(self):
return self
def __exit__(self, exc_type, exc_instance, trackback):
if exc_instance:
print("abnormalities: {}".format(exc_instance)
return False # return TrueTermination exceptions
Handling specific exceptions
class HandleValueError(object):
def __enter__(self):
return self
def __exit__(self, exc_type, exc_instance, trackback):
if not exc_type: return True
if issubclass(exc_type, ValueError):
print("deal withValueError: {}".format(exc_instance)
return False
if issubclass... If exec_type == ValueError then ValueType subclass exceptions are not handled.
It is also possible to determine whether to propagate or terminate based on the properties of the exception
Simpler syntax
import contextlib
@
def acceptable_error_codes(*codes):
try:
yield
except ShellException as exc_instance:
if exc_instance.code not in codes:
raise
pass
This is the whole content of this article.