Coverage for configzen/module_proxy.py: 78%
73 statements
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
1"""Runtime modules with attribute type validation."""
3from __future__ import annotations 1abcdefgh
5import inspect 1abcdefgh
6import sys 1abcdefgh
7import types 1abcdefgh
8from typing import Any, Generic, cast 1abcdefgh
10from configzen.typedefs import ConfigObject 1abcdefgh
12__all__ = ("ModuleProxy",) 1abcdefgh
15def _is_dunder(name: str) -> bool: 1abcdefgh
16 return ( 1abcdefgh
17 len(name) > 4 # noqa: PLR2004
18 # and name.isidentifier() -- used internally, so we don't need to check
19 and name.startswith("__")
20 and name[2] != "_"
21 and name.endswith("__")
22 )
25class ModuleProxy(types.ModuleType, Generic[ConfigObject]): 1abcdefgh
26 """
27 Proxy object that extends a runtime module with type validation.
29 Triggered via a config instance (initialization and assignment).
31 Parameters
32 ----------
33 name
34 The name of the module.
35 config
36 The configuration model to use for type validation.
37 module_namespace
38 The module namespace to wrap.
40 """
42 __config__: ConfigObject 1abcdefgh
43 __locals__: dict[str, Any] 1abcdefgh
45 def __init__( 1abcdefg
46 self,
47 name: str,
48 config: ConfigObject,
49 module_namespace: dict[str, Any] | None = None,
50 doc: str | None = None,
51 ) -> None:
52 object.__setattr__(self, "__config__", config) 1abcdefgh
53 object.__setattr__(self, "__locals__", module_namespace or {}) 1abcdefgh
54 object.__setattr__(config, "__wrapped_module__", self) 1abcdefgh
56 super().__init__(name=name, doc=doc) 1abcdefgh
58 parts = name.split(".") 1abcdefgh
59 if len(parts) > 1: 59 ↛ 61line 59 didn't jump to line 61, because the condition on line 59 was never true1abcdefgh
60 # Set the proxy module as an attribute of its parent.
61 parent = sys.modules[".".join(parts[:-1])]
62 setattr(parent, parts[-1], self)
64 # Make reusable.
65 sys.modules[name] = self 1abcdefgh
67 def __getattribute__(self, name: str) -> Any: 1abcdefgh
68 """Get an attribute of the underlying model."""
69 if _is_dunder(name): 1abcdefgh
70 return object.__getattribute__(self, name) 1abcdefgh
72 config = self.__config__ 1abcdefgh
73 try: 1abcdefgh
74 return getattr(config, name) 1abcdefgh
75 except AttributeError: 1abcdefgh
76 try: 1abcdefgh
77 return self.__locals__[name] 1abcdefgh
78 except KeyError: 1abcdefgh
79 return object.__getattribute__(self, name) 1abcdefgh
81 def __setattr__(self, key: str, value: Any) -> None: 1abcdefgh
82 """Set an attribute on the underlying model."""
83 config = self.get_config() 1abcdefgh
84 if not _is_dunder(key) and key in config.model_fields: 84 ↛ 86line 84 didn't jump to line 86, because the condition on line 84 was always true1abcdefgh
85 setattr(config, key, value) 1abcdefgh
86 self.__locals__[key] = value 1abcdefgh
88 def __repr__(self) -> str: 1abcdefgh
89 """
90 Get the string representation of the module proxy.
92 Inform the user that this is a configuration module.
93 """
94 return super().__repr__().replace("module", "configuration module", 1) 1abcdefgh
96 def get_config(self) -> ConfigObject: 1abcdefgh
97 """Get the configuration model."""
98 return self.__config__ 1abcdefgh
100 @classmethod 1abcdefgh
101 def wrap_module( 1abcdefg
102 cls,
103 module_name: str,
104 config_class: type[ConfigObject] | None = None,
105 namespace: dict[str, Any] | None = None,
106 /,
107 **values: Any,
108 ) -> ModuleProxy[ConfigObject]:
109 """
110 Wrap a module to ensure type validation.
112 Every attribute of the wrapped module that is also a field of the config will be
113 validated against it. The module will be extended with the config's attributes.
114 Assignments on the module's attributes will be propagated to the configuration
115 instance. It could be said that the module becomes a proxy for the configuration
116 once wrapped.
118 Parameters
119 ----------
120 module_name
121 The name of the module to wrap.
122 config_class
123 The config class to use for type validation.
124 namespace
125 The namespace of the module to wrap. If not provided, it will be
126 retrieved from `sys.modules`.
127 values
128 Values used to initialize the config.
130 Returns
131 -------
132 The wrapped module.
134 """
135 from configzen.config import BaseConfig 1abcdefgh
137 if namespace is None: 137 ↛ 138line 137 didn't jump to line 138, because the condition on line 137 was never true1abcdefgh
138 module_namespace = vars(sys.modules[module_name])
139 else:
140 module_namespace = namespace 1abcdefgh
142 if config_class is None: 142 ↛ 144line 142 didn't jump to line 144, because the condition on line 142 was never true1abcdefgh
144 class ConfigModule(BaseConfig):
145 __module__ = module_name
146 __annotations__ = module_namespace["__annotations__"]
147 for key in __annotations__:
148 locals()[key] = module_namespace[key]
150 config_class = cast("type[ConfigObject]", ConfigModule)
152 module_values = {} 1abcdefgh
153 field_names = frozenset( 1abcdefgh
154 field_info.validation_alias
155 or field_info.alias
156 or field_info.title
157 or field_name
158 for field_name, field_info in config_class.model_fields.items()
159 )
160 for key, value in module_namespace.items(): 1abcdefgh
161 if key in field_names: 1abcdefgh
162 module_values[key] = value 1abcdefgh
163 config = config_class.model_validate({**module_values, **values}) 1abcdefgh
165 return cls( 1abcdefgh
166 config=config,
167 module_namespace=module_namespace,
168 name=module_namespace.get("__name__") or module_name,
169 doc=module_namespace.get("__doc__"),
170 )
172 @classmethod 1abcdefgh
173 def wrap_this_module( 1abcdefg
174 cls,
175 config_class: type[ConfigObject] | None = None,
176 /,
177 **values: Any,
178 ) -> ModuleProxy[ConfigObject]:
179 """
180 Wrap the module calling this function.
182 For more information on wrapping modules, see `ModuleProxy.wrap_module()`.
184 Parameters
185 ----------
186 config_class
187 The config class to use for type validation.
188 values
189 Values used to initialize the config.
191 """
192 current_frame = inspect.currentframe() 1abcdefgh
193 if current_frame is None: 193 ↛ 194line 193 didn't jump to line 194, because the condition on line 193 was never true1abcdefgh
194 msg = "Could not get the current frame"
195 raise RuntimeError(msg)
196 frame_back = current_frame.f_back 1abcdefgh
197 if frame_back is None: 197 ↛ 198line 197 didn't jump to line 198, because the condition on line 197 was never true1abcdefgh
198 msg = "Could not get the frame back"
199 raise RuntimeError(msg)
200 return cls.wrap_module( 1abcdefgh
201 {**frame_back.f_globals, **frame_back.f_locals}["__name__"],
202 config_class,
203 {**frame_back.f_locals, **values},
204 )