Coverage for presidio_analyzer / recognizer_registry / recognizers_loader_utils.py: 97%
186 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 09:03 +0000
1from __future__ import annotations
3import inspect
4import logging
5from collections.abc import ItemsView
6from pathlib import Path
7from typing import (
8 Any,
9 ClassVar,
10 Dict,
11 Iterable,
12 List,
13 Optional,
14 Set,
15 Tuple,
16 Type,
17 Union,
18)
20import yaml
22from presidio_analyzer import EntityRecognizer, PatternRecognizer
24logger = logging.getLogger("presidio-analyzer")
27class PredefinedRecognizerNotFoundError(Exception):
28 """Exception raised when a predefined recognizer is not found."""
30 pass
33class RecognizerListLoader:
34 """A utility class that initializes recognizers based on configuration."""
36 SUPPORTED_ENTITY: ClassVar[str] = "supported_entity"
37 SUPPORTED_ENTITIES: ClassVar[str] = "supported_entities"
39 @staticmethod
40 def _get_recognizer_items(
41 recognizer_conf: Union[Dict[str, Any], str],
42 ) -> Union[dict[Any, Any], ItemsView[str, Any]]:
43 if isinstance(recognizer_conf, str):
44 return {}
45 return recognizer_conf.items()
47 @staticmethod
48 def is_recognizer_enabled(recognizer_conf: Union[Dict[str, Any], str]) -> bool:
49 """Return True if the recognizer is enabled.
51 :param recognizer_conf: The recognizer configuration.
52 """
53 return "enabled" not in recognizer_conf or recognizer_conf["enabled"]
55 @staticmethod
56 def _get_recognizer_context(
57 recognizer: Union[Dict[str, Any], str],
58 ) -> Optional[List[str]]:
59 if isinstance(recognizer, str):
60 return None
61 return recognizer.get("context", None)
63 @staticmethod
64 def _split_recognizers(
65 recognizers_conf: Union[Dict[str, Any], str],
66 ) -> Tuple[List[Union[str, Dict[str, Any]]], List[Union[str, Dict[str, Any]]]]:
67 """
68 Split the recognizer list to predefined and custom.
70 All recognizers are custom by default though
71 type: 'custom' can be mentioned as well.
72 This function supports the previous format as well.
74 :param recognizers_conf: The recognizers' configuration
75 """
77 predefined = [
78 recognizer_conf
79 for recognizer_conf in recognizers_conf
80 if isinstance(recognizer_conf, dict)
81 and ("type" in recognizer_conf and recognizer_conf["type"] == "predefined")
82 ]
83 custom = [
84 recognizer_conf
85 for recognizer_conf in recognizers_conf
86 if not isinstance(recognizer_conf, str)
87 and ("type" not in recognizer_conf or recognizer_conf["type"] == "custom")
88 ]
89 return predefined, custom
91 @staticmethod
92 def _get_recognizer_languages(
93 recognizer_conf: Union[Dict[str, Any], str],
94 supported_languages: Iterable[str],
95 ) -> List[Dict[str, Any]]:
96 """
97 Get the different language properties for each recognizer.
99 Creating a new recognizer for each supported language.
100 If language wasn't specified, create a recognizer for each supported language.
102 :param recognizer_conf: The aforementioned recognizer.
103 :return: The list of recognizers in the supported languages.
104 """
105 if (
106 isinstance(recognizer_conf, str)
107 or "supported_languages" not in recognizer_conf
108 or recognizer_conf["supported_languages"] is None
109 ):
110 return [
111 {
112 "supported_language": language,
113 "context": RecognizerListLoader._get_recognizer_context(
114 recognizer=recognizer_conf
115 ),
116 }
117 for language in supported_languages
118 ]
120 if isinstance(recognizer_conf["supported_languages"][0], str):
121 return [
122 {"supported_language": language, "context": None}
123 for language in recognizer_conf["supported_languages"]
124 ]
126 return [
127 {
128 "supported_language": language["language"],
129 "context": language.get("context", None),
130 }
131 for language in recognizer_conf["supported_languages"]
132 ]
134 @staticmethod
135 def get_recognizer_name(recognizer_conf: Union[Dict[str, Any], str]) -> str:
136 """Get the class name for recognizer instantiation.
138 Uses 'class_name' if present, otherwise 'name'.
140 Logic:
141 - If only 'name' exists: Use 'name' as both class name (for instantiation)
142 and instance name (passed to __init__)
143 - If 'class_name' exists: Use 'class_name' for instantiation and 'name'
144 as the instance name (passed to __init__)
146 :param recognizer_conf: The recognizer configuration.
147 """
148 if isinstance(recognizer_conf, str):
149 return recognizer_conf
150 class_name = recognizer_conf.get("class_name")
151 if class_name:
152 return class_name
153 return recognizer_conf["name"]
155 @staticmethod
156 def _convert_supported_entities_to_entity(conf: Dict[str, Any]) -> None:
157 if RecognizerListLoader.SUPPORTED_ENTITIES in conf:
158 supported_entities = conf.pop(RecognizerListLoader.SUPPORTED_ENTITIES)
159 if RecognizerListLoader.SUPPORTED_ENTITY not in conf and supported_entities:
160 conf[RecognizerListLoader.SUPPORTED_ENTITY] = supported_entities[0]
162 @staticmethod
163 def _is_language_supported_globally(
164 recognizer: EntityRecognizer,
165 supported_languages: Iterable[str],
166 ) -> bool:
167 if recognizer.supported_language not in supported_languages:
168 logger.warning(
169 f"Recognizer not added to registry because "
170 f"language is not supported by registry - "
171 f"{recognizer.name} supported "
172 f"languages: {recognizer.supported_language}"
173 f", registry supported languages: "
174 f"{', '.join(supported_languages)}"
175 )
176 return False
177 return True
179 @staticmethod
180 def _create_custom_recognizers(
181 recognizer_conf: Dict,
182 supported_languages: Iterable[str],
183 ) -> List[PatternRecognizer]:
184 """Create a custom recognizer for each language, based on the provided conf."""
185 # legacy recognizer (has supported_language set to a value, not None)
186 if recognizer_conf.get("supported_language"):
187 # Remove supported_languages field (plural) if present,
188 # as we're using supported_language (singular)
189 conf_copy = {
190 k: v for k, v in recognizer_conf.items() if k != "supported_languages"
191 }
193 # Transform supported_entities -> supported_entity
194 # (PatternRecognizer expects singular)
195 RecognizerListLoader._convert_supported_entities_to_entity(conf_copy)
197 return [PatternRecognizer.from_dict(conf_copy)]
199 recognizers = []
201 for supported_language in RecognizerListLoader._get_recognizer_languages(
202 recognizer_conf=recognizer_conf, supported_languages=supported_languages
203 ):
204 copied_recognizer = {
205 k: v
206 for k, v in recognizer_conf.items()
207 if k not in ["enabled", "type", "supported_languages"]
208 }
210 # Transform supported_entities -> supported_entity
211 # (PatternRecognizer expects singular)
212 RecognizerListLoader._convert_supported_entities_to_entity(
213 copied_recognizer
214 )
216 kwargs = {**copied_recognizer, **supported_language}
217 recognizers.append(PatternRecognizer.from_dict(kwargs))
219 return recognizers
221 @staticmethod
222 def get_all_existing_recognizers(
223 cls: Optional[Type[EntityRecognizer]] = None,
224 ) -> Set[Type[EntityRecognizer]]:
225 """
226 Return all subclasses of EntityRecognizer, recursively.
228 :param cls: The initial class, if None, cls=EntityRecognizer.
229 """
231 if not cls:
232 cls = EntityRecognizer
234 return set(cls.__subclasses__()).union(
235 [
236 s
237 for c in cls.__subclasses__()
238 for s in RecognizerListLoader.get_all_existing_recognizers(c)
239 ]
240 )
242 @staticmethod
243 def get_existing_recognizer_cls(recognizer_name: str) -> Type[EntityRecognizer]:
244 """
245 Get the recognizer class by name.
247 Returns the requested recognizer class out of the list of all existing
248 recognizers currently inheriting from EntityRecognizer.
249 Raises a ValueError if the recognizer is not found.
251 :param recognizer_name: The name of the recognizer.
252 """
253 all_existing_recognizers = RecognizerListLoader.get_all_existing_recognizers()
254 for recognizer in all_existing_recognizers:
255 if recognizer_name == recognizer.__name__:
256 return recognizer
258 raise PredefinedRecognizerNotFoundError(
259 f"Recognizer of name {recognizer_name} was not found in the "
260 f"list of recognizers inheriting the EntityRecognizer class"
261 )
263 @staticmethod
264 def _prepare_recognizer_kwargs(
265 recognizer_conf: Dict[str, Any],
266 language_conf: Dict[str, Any],
267 recognizer_cls: Type[EntityRecognizer],
268 ) -> Dict[str, Any]:
269 """
270 Prepare kwargs for recognizer instantiation.
272 This function adapts supported_entity/supported_entities based on the
273 recognizer class __init__ signature to avoid passing unexpected kwargs.
275 - If recognizer accepts only supported_entity (singular), convert
276 supported_entities -> supported_entity (first element).
277 - If recognizer accepts only supported_entities (plural), remove
278 supported_entity.
279 - If recognizer accepts both, keep keys as provided (after None cleanup).
280 - Filtering policy:
281 - supported_entity: kept only if explicitly accepted.
282 - supported_entities: kept if explicitly accepted or if the recognizer
283 accepts **kwargs.
284 """
285 kwargs = {**recognizer_conf, **language_conf}
287 # Cleanup: Remove provided entity arguments if they are explicitly None
288 if (
289 RecognizerListLoader.SUPPORTED_ENTITY in kwargs
290 and kwargs[RecognizerListLoader.SUPPORTED_ENTITY] is None
291 ):
292 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITY, None)
294 if (
295 RecognizerListLoader.SUPPORTED_ENTITIES in kwargs
296 and kwargs[RecognizerListLoader.SUPPORTED_ENTITIES] is None
297 ):
298 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITIES, None)
300 try:
301 params = inspect.signature(recognizer_cls.__init__).parameters
302 except (TypeError, ValueError):
303 # Drop entity-related kwargs to avoid passing unexpected arguments when the
304 # signature cannot be inspected.
305 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITY, None)
306 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITIES, None)
307 return kwargs
309 # If the recognizer accepts **kwargs, passing extra fields won't raise
310 # TypeError. Whether the recognizer uses them is up to the implementation.
311 has_var_kw = any(
312 p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()
313 )
315 accepts_supported_entity = RecognizerListLoader.SUPPORTED_ENTITY in params
316 accepts_supported_entities = RecognizerListLoader.SUPPORTED_ENTITIES in params
318 # 1. Normalize: Convert plural -> singular if needed
319 # (Only when singular is accepted and plural is NOT accepted)
320 if accepts_supported_entity and not accepts_supported_entities:
321 if RecognizerListLoader.SUPPORTED_ENTITIES in kwargs:
322 supported_entities = kwargs.get(RecognizerListLoader.SUPPORTED_ENTITIES)
324 # Use the first entity if available
325 if isinstance(supported_entities, list) and supported_entities:
326 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITIES)
327 kwargs.setdefault(
328 RecognizerListLoader.SUPPORTED_ENTITY, supported_entities[0]
329 )
331 # 2. Filter: Remove keys that are NOT in the signature
333 # For supported_entities (plural):
334 # If not explicitly accepted, remove unless **kwargs is present (compat).
335 if not accepts_supported_entities and not has_var_kw:
336 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITIES, None)
338 # Drop unsupported 'supported_entity' even for **kwargs
339 # to prevent leaking into strict parent __init__.
340 if not accepts_supported_entity:
341 kwargs.pop(RecognizerListLoader.SUPPORTED_ENTITY, None)
343 return kwargs
345 @staticmethod
346 def get(
347 recognizers: Dict[str, Any],
348 supported_languages: Iterable[str],
349 global_regex_flags: int,
350 ) -> Iterable[EntityRecognizer]:
351 """
352 Create an iterator of recognizers.
354 The recognizers are initialized according to configuration loaded previously.
355 """
356 recognizer_instances = []
357 predefined, custom = RecognizerListLoader._split_recognizers(recognizers)
359 predefined_to_exclude = {"enabled", "type", "supported_languages", "class_name"}
360 custom_to_exclude = {"enabled", "type", "class_name"}
361 for recognizer_conf in predefined:
362 for language_conf in RecognizerListLoader._get_recognizer_languages(
363 recognizer_conf=recognizer_conf, supported_languages=supported_languages
364 ):
365 if RecognizerListLoader.is_recognizer_enabled(recognizer_conf):
366 new_conf = RecognizerListLoader._filter_recognizer_fields(
367 recognizer_conf, to_exclude=predefined_to_exclude
368 )
370 recognizer_name = RecognizerListLoader.get_recognizer_name(
371 recognizer_conf=recognizer_conf
372 )
373 recognizer_cls = RecognizerListLoader.get_existing_recognizer_cls(
374 recognizer_name=recognizer_name
375 )
377 kwargs = RecognizerListLoader._prepare_recognizer_kwargs(
378 new_conf, language_conf, recognizer_cls
379 )
381 recognizer_instances.append(recognizer_cls(**kwargs))
383 for recognizer_conf in custom:
384 if RecognizerListLoader.is_recognizer_enabled(recognizer_conf):
385 new_conf = RecognizerListLoader._filter_recognizer_fields(
386 recognizer_conf, to_exclude=custom_to_exclude
387 )
388 recognizer_instances.extend(
389 RecognizerListLoader._create_custom_recognizers(
390 recognizer_conf=new_conf,
391 supported_languages=supported_languages,
392 )
393 )
395 for recognizer_conf in recognizer_instances:
396 if isinstance(recognizer_conf, PatternRecognizer):
397 recognizer_conf.global_regex_flags = global_regex_flags
399 recognizer_instances = [
400 recognizer
401 for recognizer in recognizer_instances
402 if RecognizerListLoader._is_language_supported_globally(
403 recognizer=recognizer, supported_languages=supported_languages
404 )
405 ]
407 return recognizer_instances
409 @staticmethod
410 def _filter_recognizer_fields(
411 recognizer_conf: Dict[str, Any], to_exclude: Set[str]
412 ) -> Dict[str, Any]:
413 copied_recognizer_conf = {
414 k: v
415 for k, v in RecognizerListLoader._get_recognizer_items(
416 recognizer_conf=recognizer_conf
417 )
418 if k not in to_exclude
419 }
420 return copied_recognizer_conf
423class RecognizerConfigurationLoader:
424 """A utility class that initializes recognizer registry configuration."""
426 mandatory_keys = [
427 "supported_languages",
428 "recognizers",
429 "global_regex_flags",
430 ]
432 @staticmethod
433 def _merge_configuration(
434 registry_configuration: Dict, config_from_file: Dict[str, Any]
435 ) -> Dict:
436 """
437 Add missing keys to the configuration.
439 Missing keys are added using the configuration read from file.
440 :param registry_configuration: The configuration to update.
441 :param config_from_file: The configuration coming from the conf file.
442 """
443 registry_configuration.update(
444 {
445 k: v
446 for k, v in config_from_file.items()
447 if k not in list(registry_configuration.keys())
448 }
449 )
451 # Validation is now handled by Pydantic via ConfigurationValidator
452 return registry_configuration
454 @staticmethod
455 def get(
456 conf_file: Optional[Union[Path, str]] = None,
457 registry_configuration: Optional[Dict] = None,
458 ) -> Union[Dict[str, Any]]:
459 """Get the configuration from the provided file or dict.
461 :param conf_file: The configuration file
462 to read the recognizer registry configuration from.
463 :param registry_configuration: The configuration to use.
464 """
466 if conf_file and registry_configuration:
467 raise ValueError(
468 "Either conf_file or registry_configuration should"
469 " be provided, not both."
470 )
472 configuration = {}
473 config_from_file = {}
474 use_defaults = True
476 if registry_configuration:
477 configuration = registry_configuration.copy()
478 # Check if registry_configuration has all mandatory keys
479 # Note: supported_languages is now optional,
480 # so we only check for recognizers
481 mandatory_keys_set = {"recognizers", "global_regex_flags"}
482 config_keys = set(configuration.keys())
483 if mandatory_keys_set.issubset(config_keys):
484 use_defaults = False
486 if conf_file:
487 try:
488 with open(conf_file) as file:
489 config_from_file = yaml.safe_load(file)
490 use_defaults = False
492 except OSError:
493 logger.warning(
494 f"configuration file {conf_file} not found. Using default config."
495 )
496 with open(RecognizerConfigurationLoader._get_full_conf_path()) as file:
497 config_from_file = yaml.safe_load(file)
498 use_defaults = False
500 except Exception as e:
501 raise ValueError(f"Failed to parse file {conf_file}. Error: {str(e)}")
503 # Load defaults if needed (no config provided,
504 # or registry_configuration is incomplete)
505 if use_defaults:
506 with open(RecognizerConfigurationLoader._get_full_conf_path()) as file:
507 config_from_file = yaml.safe_load(file)
509 if config_from_file and not isinstance(config_from_file, dict):
510 raise TypeError(
511 f"The configuration in file {conf_file} should be a valid YAML, "
512 f"got {type(config_from_file)}"
513 )
515 if registry_configuration and not isinstance(registry_configuration, dict):
516 raise TypeError(
517 f"Expected registry_configuration to be a dict, "
518 f"got {type(registry_configuration)}"
519 )
521 # Check if config_from_file has any invalid keys
522 # (keys that aren't mandatory or valid optional keys)
523 # If it has keys but none of them are mandatory keys,
524 # it's likely an invalid config
525 if config_from_file and conf_file:
526 config_keys = set(config_from_file.keys())
527 mandatory_keys_set = {"recognizers"} # Only recognizers is truly mandatory
529 # If config has keys but none are mandatory and it's from a conf_file,
530 # it's probably invalid - don't merge with defaults
531 if config_keys and not config_keys.intersection(mandatory_keys_set):
532 raise ValueError(
533 f"Configuration file {conf_file} does not contain any of the "
534 f"mandatory keys: {list(mandatory_keys_set)}. "
535 f"Found keys: {list(config_keys)}"
536 )
538 configuration = RecognizerConfigurationLoader._merge_configuration(
539 registry_configuration=configuration, config_from_file=config_from_file
540 )
542 return configuration
544 @staticmethod
545 def _get_full_conf_path(
546 default_conf_file: Union[Path, str] = "default_recognizers.yaml",
547 ) -> Path:
548 """Return a Path to the default conf file."""
549 return Path(Path(__file__).parent, "../conf", default_conf_file)