七个好用的装饰器
1、dispach
Python 天然支持多态,但使用 dispatch 可以让你的代码更加容易阅读。
安装:
pip install multipledispatch
使用:
>>> from multipledispatch import dispatch >>> @dispatch(int, int) ... def add(x, y): ... return x + y >>> @dispatch(object, object) ... def add(x, y): ... return "%s + %s" % (x, y) >>> add(1, 2) 3 >>> add(1, hello) 1 + hello
2、click
click 可以很方便地让你实现命令行工具。
安装:
pip install click
使用:demo2.py :
import click @click.command() @click.option(--count, default=1, help=Number of greetings.) @click.option(--name, prompt=Your name, help=The person to greet.) def hello(count, name): """Simple program that greets NAME for a total of COUNT times.""" for x in range(count): click.echo(f"Hello {name}!") if __name__ == __main__: hello()
运行结果:
❯ python demo2.py --count=3 --name=joih Hello joih! Hello joih! Hello joih! ❯ python demo2.py --count=3 Your name: somenzz Hello somenzz! Hello somenzz! Hello somenzz!
3、celery
分布式的任务队列,非 Celery 莫属。
from celery import Celery app = Celery(tasks, broker=pyamqp://guest@localhost//) @app.task def add(x, y): return x + y
4、deprecated
这个相信大家在使用别的包时都遇到过,当要下线一个老版本的函数的时候就可以使用这个装饰器。
安装:
pip install Deprecated
使用:demo4.py
from deprecated import deprecated @deprecated ("This function is deprecated, please do not use it") def func1(): pass func1()
运行效果如下:
❯ python demo4.py demo4.py:6: DeprecationWarning: Call to deprecated function (or staticmethod) func1. (This function is deprecated, please do not use it) func1()
5、deco.concurrent
安装:
pip install deco
使用 DECO 就像在 Python 程序中查找或创建两个函数一样简单。我们可以用 @concurrent 装饰需要并行运行的函数,用 @synchronized 装饰调用并行函数的函数,使用举例:
from deco import concurrent, synchronized @concurrent # We add this for the concurrent function def process_url(url, data): #Does some work which takes a while return result @synchronized # And we add this for the function which calls the concurrent function def process_data_set(data): results = {} for url in urls: results[url] = process_url(url, data) return results
6、cachetools
缓存工具
安装:
pip install cachetools
使用:
from cachetools import cached, LRUCache, TTLCache # speed up calculating Fibonacci numbers with dynamic programming @cached(cache={}) def fib(n): return n if n < 2 else fib(n - 1) + fib(n - 2) # cache least recently used Python Enhancement Proposals @cached(cache=LRUCache(maxsize=32)) def get_pep(num): url = http://www.python.org/dev/peps/pep-%04d/ % num with urllib.request.urlopen(url) as s: return s.read() # cache weather data for no longer than ten minutes @cached(cache=TTLCache(maxsize=1024, ttl=600)) def get_weather(place): return owm.weather_at_place(place).get_weather()
7、retry
重试装饰器,支持各种各样的重试需求。
安装:
pip install tenacity
使用:
import random from tenacity import retry @retry def do_something_unreliable(): if random.randint(0, 10) > 1: raise IOError("Broken sauce, everything is hosed!!!111one") else: return "Awesome sauce!" @retry(stop=stop_after_attempt(7)) def stop_after_7_attempts(): print("Stopping after 7 attempts") raise Exception @retry(stop=stop_after_delay(10)) def stop_after_10_s(): print("Stopping after 10 seconds") raise Exception @retry(stop=(stop_after_delay(10) | stop_after_attempt(5))) def stop_after_10_s_or_5_retries(): print("Stopping after 10 seconds or 5 retries") raise Exception
THE END