from Controllers.Redis.database import RedisActions def example_set_json() -> None: """Example of setting JSON data in Redis with and without expiry.""" # Example 1: Set JSON without expiry data = {"name": "John", "age": 30, "city": "New York"} keys = ["user", "profile", "123"] result = RedisActions.set_json(list_keys=keys, value=data) print("Set JSON without expiry:", result.as_dict()) # Example 2: Set JSON with expiry expiry = {"hours": 1, "minutes": 30} result = RedisActions.set_json(list_keys=keys, value=data, expires=expiry) print("Set JSON with expiry:", result.as_dict()) def example_get_json() -> None: """Example of retrieving JSON data from Redis.""" # Example 1: Get all matching keys keys = ["user", "profile", "*"] result = RedisActions.get_json(list_keys=keys) print("Get all matching JSON:", result.as_dict()) # Example 2: Get with limit result = RedisActions.get_json(list_keys=keys, limit=5) print("Get JSON with limit:", result.as_dict()) def example_get_json_iterator() -> None: """Example of using the JSON iterator for large datasets.""" keys = ["user", "profile", "*"] for row in RedisActions.get_json_iterator(list_keys=keys): print("Iterating over JSON row:", row.as_dict if isinstance(row.as_dict, dict) else row.as_dict()) # def example_delete_key() -> None: # """Example of deleting a specific key.""" # key = "user:profile:123" # result = RedisActions.delete_key(key) # print("Delete specific key:", result) # # def example_delete() -> None: # """Example of deleting multiple keys matching a pattern.""" # keys = ["user", "profile", "*"] # result = RedisActions.delete(list_keys=keys) # print("Delete multiple keys:", result) def example_refresh_ttl() -> None: """Example of refreshing TTL for a key.""" key = "user:profile:123" new_expiry = {"hours": 2, "minutes": 0} result = RedisActions.refresh_ttl(key=key, expires=new_expiry) print("Refresh TTL:", result.as_dict()) def example_key_exists() -> None: """Example of checking if a key exists.""" key = "user:profile:123" exists = RedisActions.key_exists(key) print(f"Key {key} exists:", exists) def example_resolve_expires_at() -> None: """Example of resolving expiry time for a key.""" from Controllers.Redis.base import RedisRow redis_row = RedisRow() redis_row.set_key("user:profile:123") print(redis_row.keys) expires_at = RedisActions.resolve_expires_at(redis_row) print("Resolve expires at:", expires_at) def run_all_examples() -> None: """Run all example functions to demonstrate RedisActions functionality.""" print("\n=== Redis Actions Examples ===\n") print("1. Setting JSON data:") example_set_json() print("\n2. Getting JSON data:") example_get_json() print("\n3. Using JSON iterator:") example_get_json_iterator() # print("\n4. Deleting specific key:") # example_delete_key() # # print("\n5. Deleting multiple keys:") # example_delete() print("\n6. Refreshing TTL:") example_refresh_ttl() print("\n7. Checking key existence:") example_key_exists() print("\n8. Resolving expiry time:") example_resolve_expires_at() if __name__ == "__main__": run_all_examples()