Welcome to aiocache’s documentation!¶
Installing¶
pip install aiocache
pip install aiocache[redis]
pip install aiocache[memcached]
pip install aiocache[redis,memcached]
Usage¶
Using a cache is as simple as
>>> import asyncio
>>> from aiocache import Cache
>>> cache = Cache()
>>> with asyncio.Runner() as runner:
>>> runner.run(cache.set("key", "value"))
True
>>> runner.run(cache.get("key"))
'value'
Here we are using the SimpleMemoryCache but you can use any other listed in Caches. All caches contain the same minimum interface which consists on the following functions:
add
: Only adds key/value if key does not exist. Otherwise raises ValueError.get
: Retrieve value identified by key.set
: Sets key/value.multi_get
: Retrieves multiple key/values.multi_set
: Sets multiple key/values.exists
: Returns True if key exists False otherwise.increment
: Increment the value stored in the given key.delete
: Deletes key and returns number of deleted items.clear
: Clears the items stored.raw
: Executes the specified command using the underlying client.
You can also setup cache aliases like in Django settings:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | import asyncio import redis.asyncio as redis from aiocache import caches, Cache from aiocache.serializers import StringSerializer, PickleSerializer caches.set_config({ 'default': { 'cache': "aiocache.SimpleMemoryCache", 'serializer': { 'class': "aiocache.serializers.StringSerializer" } }, 'redis_alt': { 'cache': "aiocache.RedisCache", "host": "127.0.0.1", 'port': 6379, "socket_connect_timeout": 1, 'serializer': { 'class': "aiocache.serializers.PickleSerializer" }, 'plugins': [ {'class': "aiocache.plugins.HitMissRatioPlugin"}, {'class': "aiocache.plugins.TimingPlugin"} ] } }) async def default_cache(): cache = caches.get('default') # This always returns the same instance await cache.set("key", "value") assert await cache.get("key") == "value" assert isinstance(cache, Cache.MEMORY) assert isinstance(cache.serializer, StringSerializer) async def alt_cache(): # This generates a new instance every time! You can also use # `caches.create("alt", namespace="test", etc...)` to override extra args cache = caches.create("redis_alt") await cache.set("key", "value") assert await cache.get("key") == "value" assert isinstance(cache, Cache.REDIS) assert isinstance(cache.serializer, PickleSerializer) assert len(cache.plugins) == 2 connection_args = cache.client.connection_pool.connection_kwargs assert connection_args["host"] == "127.0.0.1" assert connection_args["socket_connect_timeout"] == 1 assert connection_args["port"] == 6379 await cache.close() async def test_alias(): await default_cache() await alt_cache() cache = Cache(Cache.REDIS, client=redis.Redis()) await cache.delete("key") await cache.close() await caches.get("default").close() if __name__ == "__main__": asyncio.run(test_alias()) |
In examples folder you can check different use cases: