Merge pull request #47 from Adi3g/46-redis-module

46 redis module

.pre-commit-config.yaml

@@ -51,6 +51,7 @@ repos:
             - types-requests
             - types-pytz
             - types-python-dateutil
+            - types-redis
           exclude: ^testing/resources/
     - repo: local
       hooks:

README.md

@@ -1,4 +1,4 @@
-# Nox
+<div style="text-align:center"><img src="https://github.com/user-attachments/assets/8f5332c5-0e72-4af1-ad1f-e376d0a7877e" /></div>
 
 Nox is a Python CLI tool designed to automate and streamline various day-to-day tasks commonly performed by software engineers. It provides functionalities like JWT management, encryption/decryption, interaction with AWS S3, database operations, Docker management, and more.
 

nox/commands/redis_commands.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import click
+
+from nox.domains.redis_manager import RedisManager
+
+# Initialize Redis manager
+redis_manager = RedisManager()
+
+
+@click.group()
+def redis():
+    """Redis management commands."""
+    pass
+
+
+@click.command()
+@click.option('--key', required=True, help='Key to set in Redis')
+@click.option('--value', required=True, help='Value to set for the key')
+def set_key(key, value):
+    """Set a key in Redis."""
+    result = redis_manager.set_key(key, value)
+    click.echo(result)
+
+
+@click.command()
+@click.option('--key', required=True, help='Key to get from Redis')
+def get_key(key):
+    """Get a key from Redis."""
+    result = redis_manager.get_key(key)
+    click.echo(result)
+
+
+@click.command()
+@click.option('--key', required=True, help='Key to delete from Redis')
+def delete_key(key):
+    """Delete a key from Redis."""
+    result = redis_manager.delete_key(key)
+    click.echo(result)
+
+
+@click.command()
+@click.option('--pattern', default='*', help='Pattern to match keys (default: *)')
+def list_keys(pattern):
+    """List keys in Redis."""
+    result = redis_manager.list_keys(pattern)
+    click.echo('\n'.join(result))
+
+
+@click.command()
+def flush_db():
+    """Flush the current Redis database."""
+    result = redis_manager.flush_database()
+    click.echo(result)
+
+
+@click.command()
+def info():
+    """Get Redis server information."""
+    result = redis_manager.info()
+    click.echo(result)
+
+
+# Add commands to the redis group
+redis.add_command(set_key)
+redis.add_command(get_key)
+redis.add_command(delete_key)
+redis.add_command(list_keys)
+redis.add_command(flush_db)
+redis.add_command(info)

nox/domains/redis_manager.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from typing import Any
+
+import redis
+
+
+class RedisManager:
+    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
+        """Initialize the Redis manager."""
+        self.client = redis.Redis(host=host, port=port, db=db)
+
+    def set_key(self, key: str, value: Any) -> str:
+        """Set a key in Redis."""
+        try:
+            self.client.set(key, value)
+            return f"Key '{key}' set successfully."
+        except Exception as e:
+            return f"Error setting key '{key}': {str(e)}"
+
+    def get_key(self, key: str) -> str | None:
+        """Get a key from Redis."""
+        try:
+            value = self.client.get(key)
+            return value.decode() if value else 'Key not found.'
+        except Exception as e:
+            return f"Error getting key '{key}': {str(e)}"
+
+    def delete_key(self, key: str) -> str:
+        """Delete a key from Redis."""
+        try:
+            self.client.delete(key)
+            return f"Key '{key}' deleted successfully."
+        except Exception as e:
+            return f"Error deleting key '{key}': {str(e)}"
+
+    def list_keys(self, pattern: str = '*') -> list[str]:
+        """List keys in Redis matching a pattern."""
+        try:
+            keys = self.client.keys(pattern)
+            return [key.decode() for key in keys]
+        except Exception as e:
+            return [f"Error listing keys: {str(e)}"]
+
+    def flush_database(self) -> str:
+        """Flush the current database."""
+        try:
+            self.client.flushdb()
+            return 'Database flushed successfully.'
+        except Exception as e:
+            return f"Error flushing database: {str(e)}"
+
+    def info(self) -> str:
+        """Get Redis server info."""
+        try:
+            info = self.client.info()
+            return '\n'.join([f"{k}: {v}" for k, v in info.items()])
+        except Exception as e:
+            return f"Error retrieving server info: {str(e)}"

nox/main.py

@@ -12,6 +12,7 @@
 from nox.commands import jwt_commands
 from nox.commands import kafka_commands
 from nox.commands import net_commands
+from nox.commands import redis_commands
 from nox.commands import s3_commands
 from nox.commands import secret_commands
 from nox.commands import uuid_commands
@@ -38,6 +39,7 @@ def cli():
 cli.add_command(db_commands.db)
 cli.add_command(env_commands.env)
 cli.add_command(datetime_commands.datetime)
+cli.add_command(redis_commands.redis)
 cli.add_command(kafka_commands.kafka)
 
 if __name__ == '__main__':

requirements.txt

@@ -16,6 +16,7 @@ pycodestyle==2.12.1
 pyflakes==3.2.0
 PyJWT==2.9.0
 python-dotenv==1.0.1
+redis==5.0.8
 requests==2.32.3
 rsa==4.9
 speedtest-cli==2.1.3