Requires Python3.6+
There is a good and simple library CacheControl written for python requests library. And there is nothing similar for aiohttp. "Async CacheControl" project strives to cover this hole.
import asyncio
from acachecontrol import AsyncCache, AsyncCacheControl
async def main():
cache = AsyncCache(config={"sleep_time": 0.2})
# `AsyncCache()` with default configuration is used
# if `cache` not provided
async with AsyncCacheControl(cache=cache) as cached_sess:
async with cached_sess.get('http://example.com') as resp:
resp_text = await resp.text()
print(resp_text)
asyncio.run(main())It is possible to use any cache backend, which should implement OrderedDict interfaces: __contains__, __len__, __getitem__, __setitem__, get, pop, popitem, move_to_end:
class CustomCacheBackend():
def __init__(self):
self.item_order = []
self.storage = {}
def __contains__(self, key):
return key in self.storage
def __len__(self):
return len(self.storage)
def __getitem__(self, key):
return self.storage[key]
def __setitem__(self, key, value):
self.storage[key] = value
self.item_order.append(key)
def pop(self, key):
self.item_order.remove(key)
return self.storage.pop(key)
def move_to_end(self, key):
last_index = len(self.item_order) - 1
key_index = self.item_order.index(key)
while key_index < last_index:
self.item_order[key_index] = self.item_order[key_index+1]
key_index += 1
self.item_order[last_index] = key
def popitem(self, last=True):
key = self.item_order.pop() if last else self.item_order.pop(0)
value = self.storage.pop(key)
return valueThen you can use it in AsyncCache:
import asyncio
from acachecontrol import AsyncCache, AsyncCacheControl
async def main():
cache = AsyncCache(cache_backend=CustomCacheBackend())
async with AsyncCacheControl(cache=cache) as cached_sess:
async with cached_sess.get('http://example.com') as resp:
resp_text = await resp.text()
print(resp_text)
asyncio.run(main())Similarly, you can replace RequestContextManager (assume its implementation in module custom_implementations):
import asyncio
from acachecontrol import AsyncCache, AsyncCacheControl
from custom_implementations import CustomRequestContextManager
async def main():
async with AsyncCacheControl(request_context_manager_cls=CustomRequestContextManager) as cached_sess:
async with cached_sess.get('http://example.com') as resp:
resp_text = await resp.text()
print(resp_text)
asyncio.run(main())
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent