一个cache装饰器,允许过期,但是没有换出,没有清除
cache的必要元素
- key -> value的数据存储结构
- cache的超时时间
完整的实现cache的装饰器
from functools import wraps
import inspect
import datetime
def cache(exp=0):
def _cache(fn):
cache = {} # 定义存储cache的字典
@wraps(fn)
def wrap(*args, **kwargs):
# 拼装cache的key
key = []
names = set()
params = inspect.signature(fn).parameters
for i,arg in enumerate(args): # 位置参数
name = list(params.keys())[i]
key.append((name, arg))
names.add(name)
key.extend(kwargs.items()) #关键字参数
names.update(kwargs.keys())
for k,v in params.items():
if k not in names:
key.append((k, v.default))
key.sort(key=lambda x: x[0])
key = '&'.join(['{}={}'.format(name,arg) for name,arg in key])
# cache过期时间
if key in cache.keys(): # 判断新的数据在cache中是否存在
ret, timestamp = cache[key]
if exp == 0 or datetime.datetime.now().timestamp() - timestamp < exp:
print('cache hit')
return ret
ret = fn(*args, **kwargs)
print('cache miss')
cache[key] = (ret, datetime.datetime.now().timestamp()) # 若cache中没有刚刚传输的数据,那么则创建新的数据
return ret
return wrap
return _cache
测试结果
@cache(5)
def add(x,y=3):
return x + y
add(3,5)
上面的装饰器一定要在Python解释器中测试,不要使用开发工具来测试,因为不会不点的从新赋值函数,和执行函数,导致输出的结果都是一样的
使用参数名为key,值为value
使用带参数的装饰器实现cache的过期时间
命令分发器
通用的命令分发器,程序员可以方便的注册函数到某个命令,当用户输入某个命令时,路由到注册的函数,如果此命令没有注册函数,执行默认函数
# -*- coding: utf-8 -*-
# 命令分发例子
def command():
commands = {}
# 注册命令到函数中
def register(command):
def _register(fn):
# 在函数中若是命令存在,则抛出异常
if command in commands:
raise Exception('command {} exist'.format(command))
commands[command] = fn
return fn
return _register
# 默认值
def default_fn():
print('unkonwn command')
# 命令执行
def run():
while True:
cmd = input('enter command:') # 输入执行的命令
if cmd.strip() == 'quit': # 输入quit的时候退出
return
commands.get(cmd.strip(), default_fn)()
return register, run
# 解构注册和运行的方法
register, run = command()
# 注册函数命令到函数
@register('add')
def add():
print('lanyulei is a shuaige')
# 执行函数
run()
enter command: add
lanyulei is a shuaige
装饰器的用途
装饰器用在一种AOP的编程模型里面
监控,日志,缓存,路由,权限,参数检查,审计等等,都是常用装饰器的
名词解释:AOP 面向方面编程,针对一类问题做处理就是AOP编程,与具体的业务逻辑无关,只处理自己的负责的问题