khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241125) Testing unit written for cache (redis).

Browse Source
This commit is contained in:
khushal
2024-11-25 16:00:35 +05:30
parent b5a4f0cd52
commit bccf61459e
1 changed files with 32 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/cache/async_redis_cache_v2.py
Vendored
+32 -39
View File
@@ -406,53 +406,46 @@ class AsyncRedisCache:
if __name__ == "__main__":
import asyncio
def set_complex(x):
return {"r": x.real, "i": x.imag}
def get_complex(x):
return complex(x["r"], x["i"])
from utils_v2.string import json
async def main():
my_cache = AsyncRedisCache(
connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@redis.ditscentre.in/0",
connection_string = str(input("Paste your conn. str.: ")),
ping_counter = 100,
debug = False
debug = True
)
print("Connecting.")
connected = await my_cache.connect()
print("Success:", connected)
my_cache.add_converter(
type_name = type(2j).__name__,
set_converter_func = set_complex,
get_converter_func = get_complex
)
# Proceed only if we connected to the database successfully:
if connected:
value = await my_cache.get(key = "6b61afd0-b066-4611-9791-411a30d34624")
print("GET:", value)
print("TYP:", type(value), end = "\n\n")
value = await my_cache.get(key = str(input("Paste a key to get: ")))
print("GET:", json.to_string(value, default = str) if isinstance(value, (dict, list)) else value)
print("TYP:", type(value), end = "\n\n")
# success = await my_cache.set(
# key = "name",
# value = {"first": "John", "last": "Doe"},
# expiry = 10
# )
# print("SET:", success, end = "\n\n")
#
# value = await my_cache.get(key = "name")
# print("GET:", value)
# print("TYP:", type(value), end = "\n\n")
#
# success = await my_cache.delete(key = "name")
# print("DEL:", success, end = "\n\n")
#
# value = await my_cache.get(key = "cnt")
# print("GET:", value)
# print("TYP:", type(value), end = "\n\n")
#
# await my_cache.delete(key = "cnt")
# for _ in range(50):
# await asyncio.sleep(1.0)
# counter = await my_cache.count(key = "cnt", value = 1, expiry = 10)
# print("COUNTER:", counter)
success = await my_cache.set(
key = "name",
value = {"first": "John", "last": "Doe"},
expiry = 10
)
print("SET:", success, end = "\n\n")
value = await my_cache.get(key = "name")
print("GET:", value)
print("TYP:", type(value), end = "\n\n")
success = await my_cache.delete(key = "name")
print("DEL:", success, end = "\n\n")
print("Starting count test.")
await my_cache.delete(key = "cnt")
for _ in range(5):
print("Step:", _)
await asyncio.sleep(1.0)
counter = await my_cache.count(key = "cnt", value = 1, expiry = 10)
print("COUNTER:", counter)
asyncio.run(main())