133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Thursday, 19th Sept., 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To provide a quick way to test out Redis.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# System-level:
|
|
import os
|
|
|
|
# Utils:
|
|
from utils_v2.string import json
|
|
from utils_v2.date_time import date_time
|
|
from utils_v2.cache.async_redis_cache_v3 import AsyncRedisCache
|
|
|
|
# For async activities:
|
|
import asyncio
|
|
|
|
# Common:
|
|
from shared import constants
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
async def main():
|
|
|
|
print("Connecting...")
|
|
my_cache = AsyncRedisCache(
|
|
sentinel_json = json.from_string(input("Sentinel JSON String: ")),
|
|
ping_counter = 100,
|
|
debug = False
|
|
)
|
|
await my_cache.connect()
|
|
print("Connected!", end = "\n\n")
|
|
|
|
value = await my_cache.get(key = "name")
|
|
print("GET:", 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)
|
|
|
|
|
|
asyncio.run(main())
|