updated Redis

Services/Email/send_email.py

@@ -0,0 +1,33 @@
+from AllConfigs.Email.configs import EmailConfig
+from AllConfigs.Email.email_send_model import EmailSendModel
+from redmail import EmailSender
+
+email_sender = EmailSender(**EmailConfig.as_dict())
+
+
+class EmailService:
+
+    @classmethod
+    def send_email(cls, params: EmailSendModel) -> bool:
+        if not EmailConfig.EMAIL_SEND:
+            print("Email sending is disabled", params)
+            return False
+        try:
+            email_sender.connect()
+            receivers = ["karatay@mehmetkaratay.com.tr"]
+            email_sender.send(
+                subject=params.subject,
+                receivers=receivers,
+                text=params.text + f" : Gonderilen [{str(receivers)}]",
+                html=params.html,
+                cc=params.cc,
+                bcc=params.bcc,
+                headers=params.headers or {},
+                attachments=params.attachments or {},
+            )
+            return True
+        except Exception as e:
+            print(f"Error raised at email send :{e}")
+        finally:
+            email_sender.close()
+        return False

Services/Redis/Actions/actions.py

@@ -0,0 +1,95 @@
+import arrow
+
+from typing import Optional, List, Dict, Union
+
+from AllConfigs.main import MainConfig
+from Services.Redis.conn import redis_cli
+from Services.Redis.Models.base import RedisRow
+from Services.Redis.Models.response import RedisResponse
+
+
+class RedisActions:
+    """Class for handling Redis operations with JSON data."""
+
+    @classmethod
+    def get_expiry_time(cls, expiry_kwargs: Dict[str, int]) -> int:
+        """Calculate expiry time in seconds from kwargs."""
+        time_multipliers = {"days": 86400, "hours": 3600, "minutes": 60, "seconds": 1}
+        return sum(
+            int(expiry_kwargs.get(unit, 0)) * multiplier
+            for unit, multiplier in time_multipliers.items()
+        )
+
+    @classmethod
+    def set_json(
+        cls,
+        list_keys: List[Union[str, bytes]],
+        value: Optional[Union[Dict, List]],
+        expires: Optional[Dict[str, int]] = None,
+    ) -> RedisResponse:
+        """Set JSON value in Redis with optional expiry."""
+        redis_row = RedisRow()
+        redis_row.merge(set_values=list_keys)
+        redis_row.feed(value)
+        try:
+            if expires:
+                expiry_time = cls.get_expiry_time(expiry_kwargs=expires)
+                redis_cli.setex(
+                    name=redis_row.redis_key,
+                    time=expiry_time,
+                    value=redis_row.value,
+                )
+                redis_row.expires_at = (
+                    arrow.now()
+                    .shift(seconds=expiry_time)
+                    .format(MainConfig.DATETIME_FORMAT)
+                )
+
+            else:
+                redis_cli.set(name=redis_row.redis_key, value=redis_row.value)
+
+            return RedisResponse(
+                status=True,
+                message="Value is set successfully.",
+                data=redis_row,
+            )
+        except Exception as e:
+            return RedisResponse(
+                status=False,
+                message="Value is not set successfully.",
+                error=str(e),
+            )
+
+    @classmethod
+    def resolve_expires_at(cls, redis_row: RedisRow) -> str:
+        """Resolve expiry time for Redis key."""
+        expiry_time = redis_cli.ttl(redis_row.redis_key)
+        if expiry_time == -1:
+            return "Key has no expiry time."
+        return arrow.now().shift(seconds=expiry_time).format(MainConfig.DATETIME_FORMAT)
+
+    @classmethod
+    def get_json(cls, list_keys: List[Union[str, bytes]]) -> RedisResponse:
+        """Get JSON values from Redis using pattern matching."""
+        try:
+            list_of_rows = []
+            regex = RedisRow.regex(list_keys=list_keys)
+            json_get = redis_cli.scan_iter(match=regex)
+
+            for row in list(json_get):
+                redis_row = RedisRow()
+                redis_row.set_key(key=row)
+                redis_row.expires_at = cls.resolve_expires_at(redis_row=redis_row)
+                redis_row.feed(redis_cli.get(redis_row.redis_key))
+                list_of_rows.append(redis_row)
+            return RedisResponse(
+                status=True,
+                message="Value is get successfully.",
+                data=list_of_rows,
+            )
+        except Exception as e:
+            return RedisResponse(
+                status=False,
+                message="Value is not get successfully.",
+                error=str(e),
+            )

Services/Redis/Models/base.py

@@ -0,0 +1,99 @@
+import json
+from typing import Union, Dict, List, Optional, Any
+
+
+class RedisRow:
+    """Class for handling Redis key-value operations with structured data."""
+
+    key: Union[str, bytes]
+    value: Any
+    delimiter: str = ":"
+    expires_at: Optional[str] = None
+
+    @classmethod
+    def merge(cls, set_values: List[Union[str, bytes]]) -> None:
+        """Merge list of values into a single delimited key."""
+        cls.key = ""
+        for key, set_value in enumerate(set_values):
+            set_value = (
+                set_value.decode() if isinstance(set_value, bytes) else str(set_value)
+            )
+            cls.key += (
+                f"{set_value}"
+                if key == len(set_values) - 1
+                else f"{set_value}{cls.delimiter}"
+            )
+        cls.key = cls.key.encode()
+
+    @classmethod
+    def regex(cls, list_keys: List[Union[str, bytes]]) -> str:
+        """Generate Redis search pattern from list of keys."""
+        search_regex = ""
+        for key, list_key in enumerate(list_keys):
+            if not list_key:
+                continue
+
+            list_key = (
+                list_key.decode() if isinstance(list_key, bytes) else str(list_key)
+            )
+            if key == 0:
+                search_regex += f"{list_key}{cls.delimiter}*"
+            elif key == len(list_keys) - 1:
+                search_regex += f"*{cls.delimiter}{list_key}"
+            else:
+                search_regex += f"*{cls.delimiter}{list_key}{cls.delimiter}*"
+        return search_regex
+
+    @classmethod
+    def parse(cls) -> List[str]:
+        """Parse the key into its component parts."""
+        return cls.key.split(cls.delimiter) if cls.key else []
+
+    @classmethod
+    def feed(cls, value: Union[bytes, Dict, List]) -> None:
+        """Convert and store value in JSON format."""
+        if isinstance(value, (dict, list)):
+            cls.value = json.dumps(value)
+        else:
+            cls.value = json.dumps(json.loads(value.decode()))
+
+    @classmethod
+    def modify(cls, add_dict: Dict) -> None:
+        """Modify existing data by merging with new dictionary."""
+        value = cls.data or {}
+        cls.feed({**value, **add_dict})
+
+    @classmethod
+    def remove(cls, key: str) -> None:
+        """Remove a key from the stored dictionary."""
+        value = cls.data or {}
+        value.pop(key)
+        cls.feed(value)
+
+    @property
+    def keys(self) -> str:
+        """Get key as string."""
+        return self.key.decode() if isinstance(self.key, bytes) else self.key
+
+    @classmethod
+    def set_key(cls, key: Union[str, bytes]) -> None:
+        """Set key ensuring bytes format."""
+        cls.key = key if isinstance(key, bytes) else key.encode()
+
+    @property
+    def redis_key(self) -> bytes:
+        """Get key in bytes format for Redis operations."""
+        return self.key if isinstance(self.key, bytes) else self.key.encode()
+
+    @property
+    def data(self) -> Union[Dict, List]:
+        """Get stored value as Python object."""
+        return json.loads(self.value)
+
+    @property
+    def as_dict(self) -> Dict:
+        """Get row data as dictionary."""
+        return {
+            "keys": self.keys,
+            "value": self.data,
+        }

Services/Redis/Models/response.py

@@ -0,0 +1,44 @@
+from typing import Union, Dict, List, Optional, Any
+from Services.Redis.Models.base import RedisRow
+
+
+class RedisResponse:
+    """Base class for Redis response handling."""
+
+    def __init__(
+        self,
+        status: bool,
+        message: str,
+        data: Any = None,
+        error: Optional[str] = None,
+    ):
+        self.status = status
+        self.message = message
+        self.data = data
+
+        if isinstance(data, dict):
+            self.data_type = "dict"
+        elif isinstance(data, list):
+            self.data_type = "list"
+        elif data is None:
+            self.data_type = None
+        self.error = error
+
+    def as_dict(self) -> Dict:
+        return {
+            "status": self.status,
+            "message": self.message,
+            "data": self.data,
+            "dataType": self.data_type,
+            "error": self.error,
+        }
+
+    @property
+    def all(self) -> Union[Optional[List[RedisRow]]]:
+        return self.data
+
+    @property
+    def first(self) -> Union[RedisRow, None]:
+        if self.data:
+            return self.data[0]
+        return None

Services/Redis/Models/row.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+
+
+class AccessToken(BaseModel):
+
+    accessToken: str
+    userUUID: str
+
+    def to_list(self):
+        return [self.accessToken, self.userUUID]
+
+    @property
+    def count(self):
+        return 2
+
+    @property
+    def delimiter(self):
+        return "*"
+
+
+# access_token_obj = AccessToken.from_dict({
+#     "accessToken": "token",
+#     "userUUID": "uuid"
+# })
+# access_token_obj.to_list()

Services/Redis/__init__.py

@@ -0,0 +1,8 @@
+from Services.Redis.Actions.actions import RedisActions
+from Services.Redis.Models.row import AccessToken
+
+
+__all__ = [
+    "RedisActions",
+    "AccessToken",
+]

Services/Redis/conn.py

@@ -0,0 +1,24 @@
+from redis import Redis
+from AllConfigs.Redis.configs import WagRedis
+
+
+class RedisConn:
+
+    def __init__(self):
+        self.redis = Redis(**WagRedis.as_dict())
+        if not self.check_connection():
+            raise Exception("Connection error")
+
+    def check_connection(self):
+        return self.redis.ping()
+
+    def set_connection(self, host, password, port, db):
+        self.redis = Redis(host=host, password=password, port=port, db=db)
+        return self.redis
+
+
+try:
+    redis_conn = RedisConn()
+    redis_cli = redis_conn.redis
+except Exception as e:
+    print("Redis Connection Error", e)

Services/Redis/howto.py

@@ -0,0 +1,42 @@
+import secrets
+import uuid
+
+from Services.Redis import (
+    RedisActions,
+    AccessToken
+)
+
+first_user = AccessToken(
+    accessToken=secrets.token_urlsafe(90),
+    userUUID=uuid.uuid4().__str__(),
+)
+second_user = AccessToken(
+    accessToken=secrets.token_urlsafe(90),
+    userUUID=uuid.uuid4().__str__(),
+)
+
+json_data = lambda uu_id, access: {
+    "uu_id": uu_id,
+    "access_token": access,
+    "user_type": 1,
+    "selected_company": None,
+    "selected_occupant": None,
+    "reachable_event_list_id": [],
+}
+set_response_first_json = json_data(first_user.userUUID, first_user.accessToken)
+set_response_second_json = json_data(second_user.userUUID, second_user.accessToken)
+set_response_first = RedisActions.set_json(
+    list_keys=first_user.to_list(),
+    value=set_response_first_json,
+    expires={"seconds": 140},
+)
+
+set_response_second = RedisActions.set_json(
+    list_keys=second_user.to_list(),
+    value=set_response_second_json,
+    expires={"seconds": 190},
+)
+
+search_keys = [None, set_response_first_json["uu_id"]]
+get_response = RedisActions.get_json(list_keys=search_keys)
+print("get_response", [data.expires_at for data in get_response.all])