1  """Provides a Pydantic BaseModel for SAM configuration with dict-like access."""
  2  from pydantic import BaseModel, ValidationError
  3  from typing import Any, TypeVar, Union, get_args, get_origin
  4  from types import UnionType
  5  
  6  T = TypeVar("T", bound="SamConfigBase")
  7  
  8  
  9  class SamConfigBase(BaseModel):
 10      """
 11      A Pydantic BaseModel for SAM configuration that allows dictionary-style access
 12      for backward compatibility with components expecting dicts.
 13      Supports .get(), ['key'], and 'in' operator.
 14      """
 15  
 16      @classmethod
 17      def model_validate_and_clean(cls: type[T], obj: Any) -> T:
 18          """
 19          Validates a dictionary, first removing any keys with None values.
 20          This allows Pydantic's default values to be applied correctly when
 21          a config key is present but has no value in YAML.
 22          """
 23          if isinstance(obj, dict):
 24              cleaned_obj = {k: v for k, v in obj.items() if v is not None}
 25              return cls.model_validate(cleaned_obj)
 26          return cls.model_validate(obj)
 27  
 28      @classmethod
 29      def format_validation_error_message(cls: type[T], error: ValidationError, app_name: str | None, agent_name: str | None = None) -> str:
 30          """
 31          Formats Pydantic validation error messages into a clear, actionable format.
 32  
 33          Example output:
 34          ---- Configuration validation failed for 'my-agent-app' ----
 35  
 36             Agent Name: AgentConfig
 37  
 38          ERROR 1:
 39             Missing required field: 'namespace'
 40             Location: app_config.namespace
 41             Description: Absolute topic prefix for A2A communication (e.g., 'myorg/dev')
 42  
 43          ---- Please update your YAML configuration ----
 44          """
 45  
 46          error_lines = [
 47              f"\n---- Configuration validation failed for {app_name or 'UNKNOWN'} ----",
 48              ""
 49          ]
 50  
 51          if agent_name:
 52              error_lines.append(f"   Agent Name: {agent_name}\n")
 53  
 54          def get_nested_field_description(model_class: type[BaseModel], path: list[str | int]) -> str | None:
 55              """Recursively get field description from nested models"""
 56              if not path:
 57                  return None
 58  
 59              current_field = path[0]
 60              if str(current_field) not in model_class.model_fields:
 61                  return None
 62  
 63              field_info = model_class.model_fields[str(current_field)]
 64  
 65              if len(path) == 1:
 66                  return field_info.description
 67  
 68              annotation = field_info.annotation
 69  
 70              # Handle Optional/Union types
 71              if annotation is not None:
 72                  origin = get_origin(annotation)
 73                  if origin is Union or origin is UnionType:
 74                      types = get_args(annotation)
 75                      annotation = next((t for t in types if t is not type(None)), None)
 76                  elif origin is list:
 77                      inner_type = get_args(annotation)[0]
 78                      if len(path) > 1 and isinstance(path[1], int):
 79                          if isinstance(inner_type, type) and issubclass(inner_type, BaseModel):
 80                              return get_nested_field_description(inner_type, path[2:])
 81                          return None
 82                      annotation = inner_type
 83  
 84              if annotation is not None and isinstance(annotation, type) and issubclass(annotation, BaseModel):
 85                  return get_nested_field_description(annotation, path[1:])
 86  
 87              return None
 88  
 89  
 90          for index, err in enumerate(error.errors()):
 91              error_type = err.get('type')
 92              loc = err['loc']
 93              msg = err['msg']
 94  
 95              error_lines.append(f"ERROR {index + 1}:")
 96  
 97              absolute_path = '.'.join(str(item) for item in loc)
 98              description = get_nested_field_description(cls, list(loc))
 99              if error_type == 'missing':
100                  error_lines.extend([
101                      f"   Missing required field: '{loc[-1]}'",
102                  ])
103              else:
104                  error_lines.extend([
105                      f"   Error: {msg}",
106                  ])
107              error_lines.append(f"   Location: app_config.{absolute_path}")
108              error_lines.append(f"   Description: {description or 'UNKNOWN'}")
109              error_lines.append("")
110  
111          error_lines.append('---- Please update your YAML configuration ----')
112          return '\n'.join(error_lines) + "\n"
113  
114      def get(self, key: str, default: Any = None) -> Any:
115          """Provides dict-like .get() method."""
116          return getattr(self, key, default)
117  
118      def __getitem__(self, key: str) -> Any:
119          """Provides dict-like ['key'] access."""
120          return getattr(self, key)
121  
122      def __setitem__(self, key: str, value: Any):
123          """Provides dict-like ['key'] = value assignment."""
124          setattr(self, key, value)
125  
126      def __contains__(self, key: str) -> bool:
127          """
128          Provides dict-like 'in' support that mimics the old behavior.
129          Returns True only if the key was explicitly provided during model creation.
130          """
131          return key in self.model_fields_set
132  
133      def keys(self):
134          """Provides dict-like .keys() method."""
135          return self.model_dump().keys()
136  
137      def values(self):
138          """Provides dict-like .values() method."""
139          return self.model_dump().values()
140  
141      def items(self):
142          """Provides dict-like .items() method."""
143          return self.model_dump().items()
144  
145      def __iter__(self):
146          """Provides dict-like iteration over keys."""
147          return iter(self.model_dump())
148      
149      def pop(self, key: str, default: Any = None) -> Any:
150          """
151          Provides dict-like .pop() method.
152          Removes the attribute and returns its value, or default if not present.
153          """
154          if hasattr(self, key):
155              value = getattr(self, key)
156              # Set to None rather than deleting, as Pydantic models don't support delattr
157              setattr(self, key, None)
158              return value
159          return default