Coverage for src/mafw/processor/meta.py: 96%

20 statements  

« prev     ^ index     » next       coverage.py v7.15.0, created at 2026-07-26 09:13 +0000

1# Copyright 2025–2026 European Union 

2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu) 

3# SPDX-License-Identifier: EUPL-1.2 

4""" 

5Metaclass for the Processor framework. 

6 

7This module defines :class:`~mafw.processor.ProcessorMeta`, the metaclass responsible for 

8finalising every :class:`~mafw.processor.Processor` subclass at creation time. 

9Its duties include: 

10 

11- Ensuring parameter definitions are properly inherited and registered. 

12- Validating any declared filter schema against the allowed model constraints. 

13- Installing the super-call wrappers that guard method execution at runtime. 

14- Invoking :meth:`~mafw.processor.Processor.__post_init__` after each instance 

15 is constructed. 

16 

17.. versionadded:: 2.3 

18 Extracted from the monolithic ``processor.py`` module for improved 

19 maintainability and focused testing. 

20""" 

21 

22from __future__ import annotations 

23 

24from typing import TYPE_CHECKING, Any, cast 

25 

26from mafw.processor.parameters import _ensure_parameter_definitions, _validate_filter_schema 

27 

28if TYPE_CHECKING: 

29 from mafw.processor.base import Processor 

30 

31 

32class ProcessorMeta(type): 

33 """Metaclass that finalizes Processor subclasses before instantiation.""" 

34 

35 def __init__(cls, name: str, bases: tuple[type, ...], namespace: dict[str, Any]) -> None: 

36 """ 

37 Set up new Processor subclasses. 

38 

39 This method first checks the class for well-formed parameter definitions 

40 and filter metadata, then runs the super-call wrapping machinery described 

41 below. The wrappers are installed at class-creation time so that any later 

42 instance will be guarded before it even starts executing. 

43 """ 

44 super().__init__(name, bases, namespace) 

45 

46 # Validate and register parameter descriptors declared on the new class. 

47 _ensure_parameter_definitions(cast(type['Processor'], cls)) 

48 _validate_filter_schema(cast(type['Processor'], cls)) 

49 

50 # If the metaclass detected a deferred error during descriptor registration 

51 # (e.g. duplicate parameter names), raise it now. 

52 error = getattr(cls, '_parameter_definition_error', None) 

53 if error is not None: 

54 delattr(cls, '_parameter_definition_error') 

55 raise error 

56 

57 # For concrete Processor subclasses (not Processor itself), install wrappers 

58 # that verify super() was called in overridden lifecycle methods. 

59 if any(base.__name__ == 'Processor' for base in cls.__mro__[1:]): 

60 apply_wrappers = getattr(cls, '_apply_super_call_wrappers', None) 

61 if apply_wrappers is not None: 61 ↛ exitline 61 didn't return from function '__init__' because the condition on line 61 was always true

62 apply_wrappers() 

63 

64 def __call__(cls, *args: Any, **kwargs: Any) -> ProcessorMeta: 

65 """ 

66 Create a new Processor instance and run post-initialisation. 

67 

68 After the standard ``type.__call__`` produces the instance, this override 

69 invokes ``__post_init__`` — giving subclasses a hook that runs after 

70 ``__init__`` completes but before the caller receives the object. 

71 """ 

72 obj = type.__call__(cls, *args, **kwargs) 

73 obj.__post_init__() 

74 return cast(ProcessorMeta, obj)