chore: completes OPEN-7287 remove concrete guardrail implementations

Author: gustavocidornelas

src/openlayer/lib/guardrails/__init__.py

@@ -6,16 +6,12 @@
     GuardrailResult,
     BaseGuardrail,
     GuardrailBlockedException,
-    GuardrailRegistry,
 )
-from .pii import PIIGuardrail
 
 __all__ = [
     "GuardrailAction",
     "BlockStrategy",
-    "GuardrailResult", 
+    "GuardrailResult",
     "BaseGuardrail",
     "GuardrailBlockedException",
-    "GuardrailRegistry",
-    "PIIGuardrail",
 ]

src/openlayer/lib/guardrails/base.py

@@ -3,49 +3,61 @@
 import abc
 import enum
 import logging
-from typing import Any, Dict, List, Optional, Union, Type
+from typing import Any, Dict, Optional
 from dataclasses import dataclass
 
 logger = logging.getLogger(__name__)
 
 
 class GuardrailAction(enum.Enum):
     """Actions that a guardrail can take."""
+
     ALLOW = "allow"
     BLOCK = "block"
     MODIFY = "modify"
 
 
 class BlockStrategy(enum.Enum):
     """Strategies for handling blocked requests."""
-    RAISE_EXCEPTION = "raise_exception"  # Raise GuardrailBlockedException (breaks pipeline)
-    RETURN_EMPTY = "return_empty"        # Return empty/None response (graceful)
+
+    RAISE_EXCEPTION = (
+        "raise_exception"  # Raise GuardrailBlockedException (breaks pipeline)
+    )
+    RETURN_EMPTY = "return_empty"  # Return empty/None response (graceful)
     RETURN_ERROR_MESSAGE = "return_error_message"  # Return error message (graceful)
-    SKIP_FUNCTION = "skip_function"      # Skip function execution, return None (graceful)
+    SKIP_FUNCTION = "skip_function"  # Skip function execution, return None (graceful)
 
 
 @dataclass
 class GuardrailResult:
     """Result of applying a guardrail."""
+
     action: GuardrailAction
     modified_data: Optional[Any] = None
     metadata: Optional[Dict[str, Any]] = None
     reason: Optional[str] = None
     block_strategy: Optional[BlockStrategy] = None
     error_message: Optional[str] = None
-    
+
     def __post_init__(self):
         """Validate the result after initialization."""
         if self.action == GuardrailAction.MODIFY and self.modified_data is None:
             raise ValueError("modified_data must be provided when action is MODIFY")
         if self.action == GuardrailAction.BLOCK and self.block_strategy is None:
-            self.block_strategy = BlockStrategy.RAISE_EXCEPTION  # Default to existing behavior
+            self.block_strategy = (
+                BlockStrategy.RAISE_EXCEPTION
+            )  # Default to existing behavior
 
 
 class GuardrailBlockedException(Exception):
     """Exception raised when a guardrail blocks execution."""
-    
-    def __init__(self, guardrail_name: str, reason: str, metadata: Optional[Dict[str, Any]] = None):
+
+    def __init__(
+        self,
+        guardrail_name: str,
+        reason: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ):
         self.guardrail_name = guardrail_name
         self.reason = reason
         self.metadata = metadata or {}
@@ -54,10 +66,10 @@ def __init__(self, guardrail_name: str, reason: str, metadata: Optional[Dict[str
 
 class BaseGuardrail(abc.ABC):
     """Base class for all guardrails."""
-    
+
     def __init__(self, name: str, enabled: bool = True, **config):
         """Initialize the guardrail.
-        
+
         Args:
             name: Human-readable name for this guardrail
             enabled: Whether this guardrail is active
@@ -66,113 +78,41 @@ def __init__(self, name: str, enabled: bool = True, **config):
         self.name = name
         self.enabled = enabled
         self.config = config
-    
+
     @abc.abstractmethod
     def check_input(self, inputs: Dict[str, Any]) -> GuardrailResult:
         """Check and potentially modify function inputs.
-        
+
         Args:
             inputs: Dictionary of function inputs (parameter_name -> value)
-            
+
         Returns:
             GuardrailResult indicating the action to take
         """
         pass
-    
+
     @abc.abstractmethod
     def check_output(self, output: Any, inputs: Dict[str, Any]) -> GuardrailResult:
         """Check and potentially modify function output.
-        
+
         Args:
             output: The function's output
             inputs: Dictionary of function inputs for context
-            
+
         Returns:
             GuardrailResult indicating the action to take
         """
         pass
-    
+
     def is_enabled(self) -> bool:
         """Check if this guardrail is enabled."""
         return self.enabled
-    
+
     def get_metadata(self) -> Dict[str, Any]:
         """Get metadata about this guardrail for trace logging."""
         return {
             "name": self.name,
             "type": self.__class__.__name__,
             "enabled": self.enabled,
-            "config": self.config
+            "config": self.config,
         }
-
-
-class GuardrailRegistry:
-    """Registry for managing guardrails."""
-    
-    def __init__(self):
-        self._guardrails: Dict[str, Type[BaseGuardrail]] = {}
-    
-    def register(self, name: str, guardrail_class: Type[BaseGuardrail]):
-        """Register a guardrail class.
-        
-        Args:
-            name: Name to register the guardrail under
-            guardrail_class: The guardrail class to register
-        """
-        if not issubclass(guardrail_class, BaseGuardrail):
-            raise ValueError(f"Guardrail class must inherit from BaseGuardrail")
-        
-        self._guardrails[name] = guardrail_class
-        logger.debug(f"Registered guardrail: {name}")
-    
-    def create(self, name: str, **kwargs) -> BaseGuardrail:
-        """Create an instance of a registered guardrail.
-        
-        Args:
-            name: Name of the registered guardrail
-            **kwargs: Arguments to pass to the guardrail constructor
-            
-        Returns:
-            Instance of the guardrail
-        """
-        if name not in self._guardrails:
-            raise ValueError(f"Guardrail '{name}' not found in registry. Available: {list(self._guardrails.keys())}")
-        
-        guardrail_class = self._guardrails[name]
-        return guardrail_class(**kwargs)
-    
-    def list_available(self) -> List[str]:
-        """Get list of available guardrail names."""
-        return list(self._guardrails.keys())
-
-
-# Global registry instance
-_registry = GuardrailRegistry()
-
-
-def register_guardrail(name: str, guardrail_class: Type[BaseGuardrail]):
-    """Register a guardrail globally.
-    
-    Args:
-        name: Name to register the guardrail under
-        guardrail_class: The guardrail class to register
-    """
-    _registry.register(name, guardrail_class)
-
-
-def create_guardrail(name: str, **kwargs) -> BaseGuardrail:
-    """Create an instance of a registered guardrail.
-    
-    Args:
-        name: Name of the registered guardrail
-        **kwargs: Arguments to pass to the guardrail constructor
-        
-    Returns:
-        Instance of the guardrail
-    """
-    return _registry.create(name, **kwargs)
-
-
-def list_available_guardrails() -> List[str]:
-    """Get list of available guardrail names."""
-    return _registry.list_available()