Source code for colander_data_converter.base.common

  1from collections import OrderedDict
  2from enum import Enum
  3from typing import Dict, Any
  4
  5from pydantic import UUID4, BaseModel, model_serializer, GetCoreSchemaHandler, ValidationError, ConfigDict
  6from pydantic_core import core_schema
  7
  8type ObjectReference = UUID4
  9"""ObjectReference is an alias for UUID4, representing a unique object identifier."""
 10
 11
[docs] 12class BasePydanticEnum(Enum): 13 """Base class for creating Pydantic-compatible enums with flexible member resolution. 14 15 This class extends Python's Enum to provide seamless integration with Pydantic models. 16 It allows enum members to be resolved from various input types including codes, 17 enum members, member values, or dictionary representations. 18 19 The enum members are expected to have a `code` attribute for string-based lookup 20 and support Pydantic model validation for dictionary inputs. 21 """ 22
[docs] 23 @classmethod 24 def __get_pydantic_core_schema__(cls, _source: Any, _handler: GetCoreSchemaHandler): 25 """Define the Pydantic core schema for enum validation and serialization. 26 27 This method configures how Pydantic should validate and serialize enum instances. 28 It sets up flexible input validation that accepts multiple input formats and 29 serializes enum members using their code attribute. 30 31 Args: 32 _source (Any): The source type being processed (unused) 33 _handler (GetCoreSchemaHandler): Core schema handler from Pydantic (unused) 34 35 Returns: 36 core_schema: A Pydantic core schema that handles JSON and Python validation 37 with custom serialization to the member's code attribute 38 """ 39 # Get the member from the enum no matter what we have as input. 40 # If we fail to find a matching member, it will fail. 41 # It accepts: code, enum member and enum member value as input. 42 get_member_schema = core_schema.no_info_plain_validator_function(cls._get_member) 43 44 return core_schema.json_or_python_schema( 45 json_schema=get_member_schema, 46 python_schema=get_member_schema, 47 serialization=core_schema.plain_serializer_function_ser_schema(lambda member: member.value.code), 48 )
49 50 @classmethod 51 def _get_member(cls, input_value: Any): 52 """Resolve an enum member from various input formats. 53 54 This method provides flexible member resolution supporting multiple input types: 55 - Direct enum member instances 56 - Enum member values (the objects stored in enum members) 57 - String codes matching the member's value.code attribute 58 - Dictionary representations that can be validated as member values 59 60 Args: 61 input_value (Any): The input to resolve to an enum member. Can be: 62 - An enum member instance 63 - A member value object 64 - A string code 65 - A dictionary representation of a member value 66 67 Returns: 68 Enum member: The resolved enum member 69 70 Raises: 71 ValueError: If the input_value cannot be resolved to any enum member 72 73 Note: 74 Dictionary validation is performed for each member during iteration, 75 which may impact performance for large enums. Consider implementing 76 a _get_value_type class method in subclasses for optimization. 77 """ 78 for member in cls: 79 # If the input is already a member or is a member value, let's use it. 80 if input_value == member or input_value == member.value: 81 return member 82 83 # If not, search for the member with input_value as code. 84 if member.value.code == input_value: 85 return member 86 87 # Try to validate the input as a model, in case the user supplied a dict 88 # representing a member. Validating during each loop is suboptimal, 89 # improve this if you care about this feature. 90 # Not easy since you can't know easily the type of you member values by 91 # default. Forcing the child to implement a _get_value_type class method 92 # would solve this. 93 try: 94 model = type(member.value).model_validate(input_value) 95 except ValidationError: 96 continue 97 else: 98 # Validated successfully and matches the current member. 99 if member.value == model: 100 return member 101 102 # Raise a ValueError if our search fails for Pydantic to create its proper 103 # ValidationError. 104 raise ValueError(f"Failed to convert {input_value} to a member of {cls}")
105 106
[docs] 107class Level(BaseModel): 108 """A Pydantic model representing a hierarchical level with ordering capabilities. 109 110 This class defines a level with a code, name, ordering value, and optional description. 111 It provides comparison operators based on the ordering_value field, allowing levels 112 to be sorted and compared in a meaningful hierarchy. 113 114 Example: 115 >>> level1 = Level(code="LOW", name="Low Priority", ordering_value=10) 116 >>> level2 = Level(code="HIGH", name="High Priority", ordering_value=20) 117 >>> level1 < level2 118 True 119 >>> str(level1) 120 'Low Priority' 121 """ 122 123 model_config: ConfigDict = ConfigDict(str_strip_whitespace=True, arbitrary_types_allowed=True, from_attributes=True) 124 125 code: str 126 name: str 127 ordering_value: int 128 description: str | None = None 129
[docs] 130 @model_serializer 131 def ser_model(self) -> str: 132 return self.name
133 134 def __str__(self) -> str: 135 return self.name 136 137 def __le__(self, other: Any) -> bool: 138 if isinstance(other, Level): 139 return self.ordering_value <= other.ordering_value 140 return False 141 142 def __ge__(self, other: Any) -> bool: 143 if isinstance(other, Level): 144 return self.ordering_value >= other.ordering_value 145 return False 146 147 def __lt__(self, other: Any) -> bool: 148 if isinstance(other, Level): 149 return self.ordering_value < other.ordering_value 150 return False 151 152 def __gt__(self, other: Any) -> bool: 153 if isinstance(other, Level): 154 return self.ordering_value > other.ordering_value 155 return False 156 157 def __eq__(self, other: Any) -> bool: 158 if isinstance(other, Level): 159 return self.ordering_value == other.ordering_value and self.name == other.name 160 return False
161 162
[docs] 163class TlpPapLevel(BasePydanticEnum): 164 """Traffic Light Protocol (TLP) and Permissible Actions Protocol (PAP) classification levels. 165 166 The TLP is a set of designations used to ensure that sensitive information is shared 167 with the appropriate audience. PAP complements TLP by providing guidance on what 168 actions can be taken with the information. 169 170 Note: 171 See `FIRST TLP Standard <https://www.first.org/tlp/>`_ for complete specification. 172 173 Example: 174 >>> level = TlpPapLevel.RED 175 >>> print(level) 176 RED 177 >>> str(level) == "RED" 178 True 179 """ 180 181 RED = Level(name="RED", code="RED", ordering_value=40) 182 """Highly sensitive information, restricted to specific recipients.""" 183 184 AMBER = Level(name="AMBER", code="AMBER", ordering_value=30) 185 """Sensitive information, limited to a defined group.""" 186 187 GREEN = Level(name="GREEN", code="GREEN", ordering_value=20) 188 """Information that can be shared within the community.""" 189 190 WHITE = Level(name="WHITE", code="WHITE", ordering_value=10) 191 """Information that can be shared publicly.""" 192
[docs] 193 @classmethod 194 def by_name(cls, name: str) -> "TlpPapLevel": 195 """Retrieve a TLP/PAP level enum member by its name. 196 197 This method provides a convenient way to access enum members using their 198 string names (e.g., "RED", "AMBER", "GREEN", "WHITE"). 199 200 Args: 201 name (str): The name of the TLP/PAP level to retrieve. Must match 202 exactly one of the enum member names: "RED", "AMBER", 203 "GREEN", or "WHITE". 204 205 Returns: 206 Level: The Level object associated with the specified enum member 207 208 Raises: 209 AttributeError: If the provided name does not correspond to any 210 enum member 211 212 Example: 213 >>> level = TlpPapLevel.by_name("RED") 214 >>> print(level.name) 215 RED 216 >>> print(level.value.ordering_value) 217 40 218 """ 219 return getattr(cls, name)
220
[docs] 221 def __str__(self): 222 """Return the string representation of the TLP level. 223 224 Returns: 225 str: The TLP level value as a string 226 """ 227 return self.value.name
228 229 def __repr__(self): 230 return self.name
231 232
[docs] 233class LRUDict(OrderedDict): 234 """ 235 A dictionary with Least Recently Used (LRU) eviction policy. 236 237 This class extends OrderedDict to automatically remove the oldest items 238 when the cache exceeds a specified length. Accessing or setting an item 239 moves it to the end of the dictionary, marking it as most recently used. 240 """ 241
[docs] 242 def __init__(self, *args, cache_len: int = 4096, **kwargs): 243 """ 244 Initialize the LRUDict. 245 246 Args: 247 cache_len: Maximum number of items to keep in the cache. 248 """ 249 assert cache_len > 0 250 self.cache_len = cache_len 251 super().__init__(*args, **kwargs)
252
[docs] 253 def __setitem__(self, key, value): 254 """ 255 Set an item in the dictionary and move it to the end. 256 Evict the least recently used item if the cache exceeds its length. 257 258 Args: 259 key: The key to set. 260 value: The value to associate with the key. 261 """ 262 super().__setitem__(key, value) 263 super().move_to_end(key) 264 while len(self) > self.cache_len: 265 old_key = next(iter(self)) 266 super().__delitem__(old_key)
267
[docs] 268 def __getitem__(self, key): 269 """ 270 Retrieve an item and move it to the end as most recently used. 271 272 Args: 273 key: The key to retrieve. 274 275 Returns: 276 The value associated with the key. 277 """ 278 val = super().__getitem__(key) 279 super().move_to_end(key) 280 return val
281 282
[docs] 283class Singleton(type): 284 """Metaclass implementation of the Singleton design pattern. 285 286 This metaclass ensures that only one instance of a class can exist at any time. 287 Subsequent instantiation attempts will return the existing instance rather than 288 creating a new one. 289 290 Note: 291 The singleton instance is created lazily on first instantiation and persists 292 for the lifetime of the Python process. 293 294 Classes using this metaclass should be designed to handle reinitialization 295 gracefully, as ``__init__`` may be called multiple times on the same instance. 296 297 Example: 298 >>> class Configuration(metaclass=Singleton): 299 ... def __init__(self, value=None): 300 ... if not hasattr(self, 'initialized'): 301 ... self.value = value 302 ... self.initialized = True 303 ... 304 >>> config1 = Configuration(value=42) 305 >>> config2 = Configuration(value=99) 306 >>> print(config1 is config2) # Both variables point to the same instance 307 True 308 >>> print(config1.value) # The value from first initialization 309 42 310 """ 311 312 _instances: Dict[type, type] = {} 313
[docs] 314 def __call__(cls, *args, **kwargs): 315 """Control instance creation to ensure singleton behavior. 316 317 Args: 318 cls (type): The class being instantiated 319 *args: Positional arguments for class initialization 320 **kwargs: Keyword arguments for class initialization 321 322 Returns: 323 type: The singleton instance of the class 324 325 Note: 326 If an instance already exists, ``__init__`` will still be called with 327 the provided arguments, but no new instance is created. 328 """ 329 if cls not in cls._instances: 330 cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) 331 return cls._instances[cls]