Source code for ccat_data_transfer.settings_manager

import json
from typing import Any, Optional
from datetime import datetime, timezone
from sqlalchemy.orm import Session
from ccat_ops_db import models
from .utils import get_redis_connection




class SettingsManager:
    """Manager for handling dynamic system settings with Redis caching."""



    def __init__(self):
        self.redis = get_redis_connection()
        self.cache_prefix = "settings:"
        self.cache_ttl = 300  # 5 minutes cache TTL


    def _get_cache_key(self, key: str) -> str:
        """Get the Redis cache key for a setting."""
        return f"{self.cache_prefix}{key}"



    def get_setting(self, session: Session, key: str, default: Any = None) -> Any:
        """
        Get a setting value, first checking Redis cache, then database.

        Parameters
        ----------
        session : Session
            SQLAlchemy database session
        key : str
            Setting key to retrieve
        default : Any, optional
            Default value if setting not found

        Returns
        -------
        Any
            The setting value
        """
        # Try Redis cache first
        cache_key = self._get_cache_key(key)
        cached_value = self.redis.get(cache_key)

        if cached_value is not None:
            return json.loads(cached_value)

        # If not in cache, get from database
        setting = session.query(models.SystemSettings).filter_by(key=key).first()

        if setting is None:
            return default

        # Cache the value
        self.redis.setex(cache_key, self.cache_ttl, json.dumps(setting.value))

        return setting.value




    def set_setting(
        self,
        session: Session,
        key: str,
        value: Any,
        description: Optional[str] = None,
        updated_by: Optional[str] = None,
    ) -> None:
        """
        Set a setting value in both database and Redis cache.

        Parameters
        ----------
        session : Session
            SQLAlchemy database session
        key : str
            Setting key to set
        value : Any
            Value to set
        description : str, optional
            Description of the setting
        updated_by : str, optional
            Username of who updated the setting
        """
        # Update or create in database
        setting = session.query(models.SystemSettings).filter_by(key=key).first()

        if setting is None:
            setting = models.SystemSettings(
                key=key, value=value, description=description, updated_by=updated_by
            )
            session.add(setting)
        else:
            setting.value = value
            setting.description = description
            setting.updated_by = updated_by
            setting.updated_at = datetime.now(timezone.utc)

        session.commit()

        # Update Redis cache
        cache_key = self._get_cache_key(key)
        self.redis.setex(cache_key, self.cache_ttl, json.dumps(value))

        # Publish update event
        self.redis.publish(
            "settings:updates",
            json.dumps(
                {
                    "key": key,
                    "value": value,
                    "updated_at": setting.updated_at.isoformat(),
                    "updated_by": updated_by,
                }
            ),
        )




    def delete_setting(self, session: Session, key: str) -> None:
        """
        Delete a setting from both database and Redis cache.

        Parameters
        ----------
        session : Session
            SQLAlchemy database session
        key : str
            Setting key to delete
        """
        # Delete from database
        setting = session.query(models.SystemSettings).filter_by(key=key).first()
        if setting:
            session.delete(setting)
            session.commit()

        # Delete from Redis cache
        cache_key = self._get_cache_key(key)
        self.redis.delete(cache_key)

        # Publish deletion event
        self.redis.publish(
            "settings:updates", json.dumps({"key": key, "action": "deleted"})
        )




# Create a singleton instance
settings_manager = SettingsManager()