如何在 class 属性中缓存 API 数据?
How can I cache API data in a class attribute?
我想要一个 class,类似于下面的那个,我可以在其中使用 属性 从 API 访问数据。不过,我希望能够将数据缓存一段时间,这样我就可以快速访问该属性而不受速率限制。实现此功能的最简洁方法是什么?
from requests import get
class GitHubUser:
def __init__(self, user):
self.user = user
print("{user} has {repos} public repos!".format(
user=user, repos=self.public_repos
))
@property
def _api_data(self):
return get(
"https://api.github.com/users/{user}".format(user=self.user)
).json()
@property
def public_repos(self):
return self._api_data["public_repos"]
这里有一种让它看起来很整洁的方法(它可能看起来过于复杂,但实际上并非如此):
from uuid import uuid4
from datetime import datetime
class cached_descriptor(object):
def __init__(self, func, timeout):
self.__doc__ = getattr(func, '__doc__')
self.func = func
self.uuid = str(uuid.uuid4())
self.timeout = timeout
def __get__(self, obj, cls):
if obj is None:
return self
if not hasattr(obj, '_cache'):
obj._cache = {}
if self.uuid not in obj._cache:
obj._cache[self.uuid] = []
data = obj._cache[self.uuid]
now = datetime.now()
if not data or (now - data[1]).total_seconds() > self.timeout:
obj._cache[self.uuid] = (self.func(obj), now)
return obj._cache[self.uuid][0]
class cached_property(object):
def __init__(self, timeout):
self.timeout = timeout
def __call__(self, func):
return cached_descriptor(func, self.timeout)
分解:
cached_property
是一个装饰器工厂,它在几秒钟内接受一个 timeout
参数
cached_descriptor
是一个只读描述符,将缓存值和对象本身的时间戳存储在_cache
字典中,在一个随机生成的uuid下,以避免多个缓存属性之间的冲突
- 第一次调用时,将始终调用该函数
- 每次下一次调用,只有超过超时才会调用该函数
这是一个如何工作的例子:
import time
class A(object):
def __init__(self):
self.n_f = self.n_g = 0
@cached_property(0.1)
def f(self):
self.n_f += 1
print('calling f', self.n_f)
return self.n_f
@cached_property(0.5)
def g(self):
self.n_g += 1
print('calling g', self.n_g)
return self.n_g
a = A()
print('f', a.f)
print('g', a.g)
print('f', a.f)
print('g', a.g)
print('sleep 0.2')
time.sleep(0.2)
print('f', a.f)
print('g', a.g)
print('sleep 0.4')
time.sleep(0.4)
print('f', a.f)
print('g', a.g)
输出
calling f 1
f 1
calling g 1
g 1
f 1
g 1
sleep 0.2
calling f 2
f 2
g 1
sleep 0.4
calling f 3
f 3
calling g 2
g 2
我想要一个 class,类似于下面的那个,我可以在其中使用 属性 从 API 访问数据。不过,我希望能够将数据缓存一段时间,这样我就可以快速访问该属性而不受速率限制。实现此功能的最简洁方法是什么?
from requests import get
class GitHubUser:
def __init__(self, user):
self.user = user
print("{user} has {repos} public repos!".format(
user=user, repos=self.public_repos
))
@property
def _api_data(self):
return get(
"https://api.github.com/users/{user}".format(user=self.user)
).json()
@property
def public_repos(self):
return self._api_data["public_repos"]
这里有一种让它看起来很整洁的方法(它可能看起来过于复杂,但实际上并非如此):
from uuid import uuid4
from datetime import datetime
class cached_descriptor(object):
def __init__(self, func, timeout):
self.__doc__ = getattr(func, '__doc__')
self.func = func
self.uuid = str(uuid.uuid4())
self.timeout = timeout
def __get__(self, obj, cls):
if obj is None:
return self
if not hasattr(obj, '_cache'):
obj._cache = {}
if self.uuid not in obj._cache:
obj._cache[self.uuid] = []
data = obj._cache[self.uuid]
now = datetime.now()
if not data or (now - data[1]).total_seconds() > self.timeout:
obj._cache[self.uuid] = (self.func(obj), now)
return obj._cache[self.uuid][0]
class cached_property(object):
def __init__(self, timeout):
self.timeout = timeout
def __call__(self, func):
return cached_descriptor(func, self.timeout)
分解:
cached_property
是一个装饰器工厂,它在几秒钟内接受一个timeout
参数cached_descriptor
是一个只读描述符,将缓存值和对象本身的时间戳存储在_cache
字典中,在一个随机生成的uuid下,以避免多个缓存属性之间的冲突- 第一次调用时,将始终调用该函数
- 每次下一次调用,只有超过超时才会调用该函数
这是一个如何工作的例子:
import time
class A(object):
def __init__(self):
self.n_f = self.n_g = 0
@cached_property(0.1)
def f(self):
self.n_f += 1
print('calling f', self.n_f)
return self.n_f
@cached_property(0.5)
def g(self):
self.n_g += 1
print('calling g', self.n_g)
return self.n_g
a = A()
print('f', a.f)
print('g', a.g)
print('f', a.f)
print('g', a.g)
print('sleep 0.2')
time.sleep(0.2)
print('f', a.f)
print('g', a.g)
print('sleep 0.4')
time.sleep(0.4)
print('f', a.f)
print('g', a.g)
输出
calling f 1
f 1
calling g 1
g 1
f 1
g 1
sleep 0.2
calling f 2
f 2
g 1
sleep 0.4
calling f 3
f 3
calling g 2
g 2