40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
|
import json
|
||
|
|
import redis.asyncio as redis_async
|
||
|
|
|
||
|
|
|
||
|
|
class Cache:
|
||
|
|
def __init__(self, options):
|
||
|
|
if isinstance(options, str):
|
||
|
|
self.redis = redis_async.from_url(options, decode_responses=True)
|
||
|
|
|
||
|
|
elif isinstance(options, dict):
|
||
|
|
host = options.get("host", "localhost")
|
||
|
|
port = options.get("port", 6379)
|
||
|
|
db = options.get("db", 0)
|
||
|
|
password = options.get("password")
|
||
|
|
|
||
|
|
if password:
|
||
|
|
url = f"redis://:{password}@{host}:{port}/{db}"
|
||
|
|
else:
|
||
|
|
url = f"redis://{host}:{port}/{db}"
|
||
|
|
|
||
|
|
self.redis = redis_async.from_url(url, decode_responses=True)
|
||
|
|
|
||
|
|
else:
|
||
|
|
raise TypeError("options must be dict or redis url string")
|
||
|
|
|
||
|
|
async def set(self, key, value):
|
||
|
|
try:
|
||
|
|
await self.redis.set(key, value)
|
||
|
|
except Exception as err:
|
||
|
|
raise err
|
||
|
|
|
||
|
|
async def get(self, key):
|
||
|
|
try:
|
||
|
|
data = await self.redis.get(key)
|
||
|
|
return json.loads(data)
|
||
|
|
except Exception as err:
|
||
|
|
raise err
|
||
|
|
|
||
|
|
|
||
|
|
__all__ = ["Cache"]
|