cengal.code_flow_control.args_manager.versions.v_0.args_manager
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18__all__ = ['EntityWithExtendableArgs', 'ArgsManagerMixin', 'EntityArgsHolder', 'EAH', 'EntityArgsHolderExplicit', 'EAHE', 'ExtendKwargsManager', 19 'EKwargs', 'ExtendArgsManager', 'EArgs', 'ArgsManager', 'merge_func_args', 'interested_args_to_kwargs', 'func_args_to_kwargs', 20 'number_of_provided_args', 'args_kwargs', 'args_kwargs_to_str', 'ArgsKwargs', 'AK', 'prepare_arguments_positions', 'UnknownArgumentError', 21 'find_arg_position_and_value', 'try_find_arg_position_and_value'] 22 23from enum import Enum 24from typing import Any, Dict, Type, Callable, Union, Optional, Sequence, Tuple, List 25import inspect 26import copy 27 28""" 29Module Docstring 30Docstrings: http://www.python.org/dev/peps/pep-0257/ 31""" 32 33__author__ = "ButenkoMS <gtalk@butenkoms.space>" 34__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 35__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 36__license__ = "Apache License, Version 2.0" 37__version__ = "4.4.1" 38__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 39__email__ = "gtalk@butenkoms.space" 40# __status__ = "Prototype" 41__status__ = "Development" 42# __status__ = "Production" 43 44 45EntityWithExtendableArgs = Union[Type, Callable] 46 47 48class ArgsManagerMixin: 49 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 50 raise NotImplementedError 51 52 53class EntityArgsHolder: 54 def __init__(self, entity: EntityWithExtendableArgs, *args, **kwargs): 55 self.entity: EntityWithExtendableArgs = entity 56 self.args: Tuple = args 57 self.kwargs: Dict = kwargs 58 59 def __call__(self) -> Any: 60 return self.entity(*self.args, **self.kwargs) 61 62 def args_kwargs(self) -> Tuple[Tuple, Dict]: 63 return self.args, self.kwargs 64 65 def entity_args_kwargs(self) -> Tuple[EntityWithExtendableArgs, Tuple, Dict]: 66 return self.entity, self.args, self.kwargs 67 68 69EAH = EntityArgsHolder 70 71 72class EntityArgsHolderExplicit(EntityArgsHolder): 73 def __init__(self, entity: EntityWithExtendableArgs, args, kwargs): 74 super().__init__(entity) 75 self.args: Tuple = args 76 self.kwargs: Dict = kwargs 77 78 79EAHE = EntityArgsHolderExplicit 80 81 82class ExtendKwargsManager(ArgsManagerMixin): 83 """ 84 Usage: 85 am = ArgsManager( 86 EKwargs({'a': 'hello', 'next': 'world}), 87 EKwargs(a='hello', b='world') 88 ) 89 """ 90 91 def __init__(self, *args, **kwargs): 92 self.kwargs = kwargs 93 if not self.kwargs: 94 self.kwargs = args[0] 95 if not isinstance(self.kwargs, Dict): 96 raise RuntimeError('Wrong parameters') 97 98 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 99 kwargs.update(self.kwargs) 100 return args, kwargs 101 102 103EKwargs = ExtendKwargsManager 104 105 106class ExtendArgsManager(ArgsManagerMixin): 107 """ 108 Usage: 109 def my_func(first, second, third, fourth, a, b, c): 110 ... 111 112 am = ArgsManager( 113 EArgs(first, second, a='hello', b='world'), 114 ) 115 am(my_func, third, fourth, c='!') 116 117 am = ArgsManager( 118 EArgs(first, second, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.prefix), 119 ) 120 am(my_func, third, fourth, c='!') 121 122 am = ArgsManager( 123 EArgs(third, fourth, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.suffix), 124 ) 125 am(my_func, first, second, c='!') 126 127 am = ArgsManager( 128 EArgs(first, second, third, fourth, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.manager), 129 ) 130 am(my_func, c='!') 131 # Or: 132 # am(my_func, ignored, ignored, ignored, ignored, c='!') 133 134 am = ArgsManager( 135 EArgs(a='hello', b='world').args_state(ExtendArgsManager.ArgsState.original), 136 ) 137 # Or: 138 # am = ArgsManager( 139 # EArgs(ignored, ignored, ignored, ignored, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.original), 140 # ) 141 am(my_func, first, second, third, fourth, c='!') 142 """ 143 class ArgsState(Enum): 144 original = 0 145 prefix = 1 146 suffix = 2 147 manager = 3 148 149 def __init__(self, *args, **kwargs): 150 self.args = args 151 self._args_state: 'ExtendArgsManager.ArgsState' = ExtendArgsManager.ArgsState.prefix 152 self.kwargs = kwargs or dict() 153 if not isinstance(self.kwargs, Dict): 154 raise RuntimeError('Wrong parameters') 155 156 def args_state(self, args_state: 'ExtendArgsManager.ArgsState') -> 'ExtendArgsManager': 157 self._args_state = args_state 158 return self 159 160 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 161 if ExtendArgsManager.ArgsState.original == self._args_state: 162 pass 163 elif ExtendArgsManager.ArgsState.prefix == self._args_state: 164 args = tuple(list(self.args) + list(args)) 165 elif ExtendArgsManager.ArgsState.suffix == self._args_state: 166 args = tuple(list(args) + list(self.args)) 167 elif ExtendArgsManager.ArgsState.manager == self._args_state: 168 args = tuple(self.args) 169 kwargs.update(self.kwargs) 170 return args, kwargs 171 172 173EArgs = ExtendArgsManager 174 175 176class ArgsManager(ArgsManagerMixin): 177 """ 178 Usage: 179 am = ArgsManager( 180 EKwargs({'a': 'hello', 'next': 'world}), 181 EKwargs(a='hello', b='world') 182 ) 183 184 Example: 185 class Item(Enum): 186 Div = 0 187 Button = 1 188 189 def html(item, color, size, step=None, length=None, strength=None) -> Any: 190 ... 191 192 am = ArgsManager( 193 EKwargs({'size': 12, 'step': 'one'}), 194 EKwargs(color='green', strength=24) 195 ) 196 197 page = list() 198 page.append(am(html, Item.Div)) 199 page.append(am(html, Item.Button)) 200 # The same as: 201 # page = list() 202 # page.append(html(Item.Div, 'green', 12, 'one', strength=24)) 203 # page.append(html(Item.Button, 'green', 12, 'one', strength=24)) 204 """ 205 def __init__(self, *args): 206 self.managers = list(args) 207 self.one_shot_managers = list() 208 self.interceptors = list() 209 self.one_shot_interceptors = list() 210 211 def append(self, manager) -> 'ArgsManager': 212 self.managers.append(manager) 213 return self 214 215 def append_one_shot(self, manager) -> 'ArgsManager': 216 self.one_shot_managers.append(manager) 217 return self 218 219 def add_interceptor(self, interceptor: Callable) -> 'ArgsManager': 220 self.interceptors.append(interceptor) 221 return self 222 223 def add_interceptor_one_shot(self, interceptor: Callable) -> 'ArgsManager': 224 self.one_shot_interceptors.append(interceptor) 225 return self 226 227 def callable(self, entity: EntityWithExtendableArgs) -> Callable: 228 def callable_entity(*args, **kwargs) -> Any: 229 original_args: Tuple[Tuple, Dict] = (copy.copy(args), copy.copy(kwargs)) 230 for manager in self.managers: 231 args, kwargs = manager(entity, *args, **kwargs) 232 233 one_shot_managers_buff = self.one_shot_managers 234 self.one_shot_managers = type(one_shot_managers_buff)() 235 for manager in one_shot_managers_buff: 236 args, kwargs = manager(entity, *args, **kwargs) 237 238 resulting_args: Tuple[Tuple, Dict] = (args, kwargs) 239 instance = entity(*args, **kwargs) 240 241 for interceptor in self.interceptors: 242 interceptor(instance, entity, original_args, resulting_args) 243 244 one_shot_interceptors_buff = self.one_shot_interceptors 245 self.one_shot_interceptors = type(one_shot_interceptors_buff)() 246 for interceptor in one_shot_interceptors_buff: 247 interceptor(instance, entity, original_args, resulting_args) 248 249 return instance 250 return callable_entity 251 252 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 253 original_args: Tuple[Tuple, Dict] = (copy.copy(args), copy.copy(kwargs)) 254 for manager in self.managers: 255 args, kwargs = manager(entity, *args, **kwargs) 256 257 one_shot_managers_buff = self.one_shot_managers 258 self.one_shot_managers = type(one_shot_managers_buff)() 259 for manager in one_shot_managers_buff: 260 args, kwargs = manager(entity, *args, **kwargs) 261 262 resulting_args: Tuple[Tuple, Dict] = (args, kwargs) 263 instance = entity(*args, **kwargs) 264 for interceptor in self.interceptors: 265 interceptor(instance, entity, original_args, resulting_args) 266 267 return instance 268 269 270def merge_func_args(func_list: Sequence[Callable]) -> Tuple: 271 args: List = list() 272 default_args: List = list() 273 for func in func_list: 274 is_method = None 275 if inspect.ismethod(func): 276 is_method = True 277 elif callable(func): 278 is_method = False 279 else: 280 raise RuntimeError(f'Is not callable: {repr(func)}') 281 varnames = func.__code__.co_varnames 282 if is_method: 283 if len(varnames) > 0: 284 if 'self' == varnames[0]: 285 varnames = varnames[1:] 286 spec = inspect.getfullargspec(func) 287 # print(f'ArgSpec:') 288 # pprint(spec) 289 290 if spec.defaults is None: 291 args.extend(varnames) 292 else: 293 defaults_len = len(spec.defaults) 294 args.extend(varnames[:-defaults_len]) 295 default_args.extend(varnames[-defaults_len:]) 296 # print(f'Args:') 297 # pprint(args) 298 # print(f'Default args:') 299 # pprint(default_args) 300 return args + default_args 301 302 303def interested_args_to_kwargs(interested_args, args, kwargs): 304 kwargs = copy.copy(kwargs) 305 absent_args = list() 306 for arg in interested_args: 307 if arg not in kwargs: 308 absent_args.append(arg) 309 kwargs.update(zip(absent_args, args)) 310 return kwargs 311 312 313def func_args_to_kwargs(func, args, kwargs): 314 kwargs = copy.copy(kwargs) 315 interested_names = func.__code__.co_varnames 316 is_method = None 317 if inspect.ismethod(func): 318 is_method = True 319 elif callable(func): 320 is_method = False 321 else: 322 raise RuntimeError(f'Is not callable: {repr(func)}') 323 324 kwargs.update(zip(interested_names, args)) 325 not_needed_names = set(kwargs) - set(interested_names) 326 for name in not_needed_names: 327 kwargs.pop(name, None) 328 if is_method: 329 kwargs.pop('self', None) 330 return kwargs 331 332 333def number_of_provided_args(args, kwargs): 334 return len(args) + len(kwargs) 335 336 337def args_kwargs(*args, **kwargs) -> Tuple[Tuple, Dict]: 338 return args, kwargs 339 340 341def args_kwargs_to_str(args, kwargs) -> str: 342 if args: 343 args_str = ', '.join([f'{arg}' for arg in args]) 344 else: 345 args_str = str() 346 347 if kwargs: 348 kwargs_str = ', '.join([f'{key}={value}' for key, value in kwargs.items()]) 349 else: 350 kwargs_str = str() 351 352 if kwargs_str: 353 return f'{args_str}, {kwargs_str}' 354 else: 355 return args_str 356 357 358class ArgsKwargs: 359 def __init__(self, *args, **kwargs) -> None: 360 self.args: Tuple = args 361 self.kwargs: Dict = kwargs 362 363 def __call__(self) -> Any: 364 return self.args, self.kwargs 365 366 367AK = ArgsKwargs 368 369 370def prepare_arguments_positions(positional: Sequence[str], keyword_only: Sequence[str]) -> Dict[str, Optional[int]]: 371 positions: Dict[str, Optional[int]] = dict() 372 for index, name in enumerate(positional): 373 positions[name] = index 374 375 for index, name in enumerate(keyword_only): 376 positions[name] = None 377 378 return positions 379 380 381class UnknownArgumentError(Exception): 382 pass 383 384 385def find_arg_position_and_value(arg_name: str, positions: Dict[str, Optional[int]], args: Tuple, kwargs: Dict) -> Tuple[bool, Optional[int], Any]: 386 387 """ 388 Example: 389 from cengal.introspection.inspect import func_param_names, CodeParamNames 390 from cengal.code_flow_control.args_manager import prepare_arguments_positions, find_arg_position_and_value, UnknownArgumentError 391 392 def wrapper(arg_name: str = 'b', *args, **kwargs): 393 def my_func(a, b, *, c, d): 394 ... 395 396 params: CodeParamNames = func_param_names(my_func) 397 positoins: Dict[str, Optional[int]] = prepare_arguments_positions(params.positional, params.keyword_only) 398 found, pos, value = find_arg_position_and_value(arg_name, positoins) 399 if found: 400 print(f'Value of <{arg_name}> : {value}') 401 402 if pos is not None: 403 print(f'<{arg_name}> found as a positional argument at position {pos}') 404 405 return my_func(*args, **kwargs) 406 407 wrapper('a') 408 wrapper('a', 1, 2, 3, 4) 409 wrapper('d') 410 wrapper('d', 1, 2, 3, 4) 411 try: 412 wrapper('asdf') 413 except UnknownArgumentError as ex: 414 print('<{ex.args[1]}> is not a valid argument for my_func(). Valid arguments are: {ex.args[2]}') 415 raise 416 417 Args: 418 arg_name (str): _description_ 419 positions (Dict[str, Optional[int]]): _description_ 420 args (Tuple): _description_ 421 kwargs (Dict): _description_ 422 423 Raises: 424 UnknownArgumentError: _description_ 425 426 Returns: 427 Tuple[bool, bool, Optional[int], Any]: _description_ 428 """ 429 found: bool = False 430 pos = None 431 value = None 432 433 original_arg_pos: Optional[int] = None 434 try: 435 original_arg_pos = positions[arg_name] 436 except KeyError: 437 valid_arguments = positions.keys() 438 raise UnknownArgumentError(f'<{arg_name}> is not in {valid_arguments}', arg_name, valid_arguments) 439 440 if original_arg_pos is None: 441 found_in_args = False 442 else: 443 if len(args) > original_arg_pos: 444 found_in_args = True 445 else: 446 found_in_args = False 447 448 if found_in_args: 449 pos = original_arg_pos 450 value = args[pos] 451 found = True 452 else: 453 try: 454 value = kwargs.get(arg_name, None) 455 found = True 456 except KeyError: 457 pass 458 459 return found, pos, value 460 461 462def try_find_arg_position_and_value(arg_name: str, positions: Dict[str, Optional[int]], args: Tuple, kwargs: Dict) -> Tuple[bool, bool, Optional[int], Any]: 463 464 """ 465 Example: 466 from cengal.introspection.inspect import func_param_names, CodeParamNames 467 from cengal.code_flow_control.args_manager import prepare_arguments_positions, try_find_arg_position_and_value 468 469 def wrapper(arg_name: str = 'b', *args, **kwargs): 470 def my_func(a, b, *, c, d): 471 ... 472 473 params: CodeParamNames = func_param_names(my_func) 474 positoins: Dict[str, Optional[int]] = find_entity_arguments_positions(params.positional, params.keyword_only) 475 valid, found, pos, value = find_arg_position_and_value(arg_name, positoins) 476 if valid: 477 if found: 478 print(f'Value of <{arg_name}> : {value}') 479 480 if pos is not None: 481 print(f'<{arg_name}> found as a positional argument at position {pos}') 482 else: 483 print('<{arg_name}> is not a valid argument for my_func(). Valid arguments are: {positoins.keys()}') 484 485 return my_func(*args, **kwargs) 486 487 wrapper('a') 488 wrapper('a', 1, 2, 3, 4) 489 wrapper('d') 490 wrapper('d', 1, 2, 3, 4) 491 wrapper('asdf') 492 493 Args: 494 arg_name (str): _description_ 495 positions (Dict[str, Optional[int]]): _description_ 496 args (Tuple): _description_ 497 kwargs (Dict): _description_ 498 499 Returns: 500 Tuple[bool, bool, Optional[int], Any]: _description_ 501 """ 502 valid: bool = False 503 found: bool = False 504 pos = None 505 value = None 506 507 original_arg_pos: Optional[int] = None 508 try: 509 original_arg_pos = positions[arg_name] 510 valid = True 511 except KeyError: 512 pass 513 514 if original_arg_pos is None: 515 found_in_args = False 516 else: 517 if len(args) > original_arg_pos: 518 found_in_args = True 519 else: 520 found_in_args = False 521 522 if found_in_args: 523 pos = original_arg_pos 524 value = args[pos] 525 found = True 526 else: 527 try: 528 value = kwargs.get(arg_name, None) 529 found = True 530 except KeyError: 531 pass 532 533 return valid, found, pos, value
54class EntityArgsHolder: 55 def __init__(self, entity: EntityWithExtendableArgs, *args, **kwargs): 56 self.entity: EntityWithExtendableArgs = entity 57 self.args: Tuple = args 58 self.kwargs: Dict = kwargs 59 60 def __call__(self) -> Any: 61 return self.entity(*self.args, **self.kwargs) 62 63 def args_kwargs(self) -> Tuple[Tuple, Dict]: 64 return self.args, self.kwargs 65 66 def entity_args_kwargs(self) -> Tuple[EntityWithExtendableArgs, Tuple, Dict]: 67 return self.entity, self.args, self.kwargs
73class EntityArgsHolderExplicit(EntityArgsHolder): 74 def __init__(self, entity: EntityWithExtendableArgs, args, kwargs): 75 super().__init__(entity) 76 self.args: Tuple = args 77 self.kwargs: Dict = kwargs
Inherited Members
83class ExtendKwargsManager(ArgsManagerMixin): 84 """ 85 Usage: 86 am = ArgsManager( 87 EKwargs({'a': 'hello', 'next': 'world}), 88 EKwargs(a='hello', b='world') 89 ) 90 """ 91 92 def __init__(self, *args, **kwargs): 93 self.kwargs = kwargs 94 if not self.kwargs: 95 self.kwargs = args[0] 96 if not isinstance(self.kwargs, Dict): 97 raise RuntimeError('Wrong parameters') 98 99 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 100 kwargs.update(self.kwargs) 101 return args, kwargs
Usage: am = ArgsManager( EKwargs({'a': 'hello', 'next': 'world}), EKwargs(a='hello', b='world') )
107class ExtendArgsManager(ArgsManagerMixin): 108 """ 109 Usage: 110 def my_func(first, second, third, fourth, a, b, c): 111 ... 112 113 am = ArgsManager( 114 EArgs(first, second, a='hello', b='world'), 115 ) 116 am(my_func, third, fourth, c='!') 117 118 am = ArgsManager( 119 EArgs(first, second, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.prefix), 120 ) 121 am(my_func, third, fourth, c='!') 122 123 am = ArgsManager( 124 EArgs(third, fourth, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.suffix), 125 ) 126 am(my_func, first, second, c='!') 127 128 am = ArgsManager( 129 EArgs(first, second, third, fourth, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.manager), 130 ) 131 am(my_func, c='!') 132 # Or: 133 # am(my_func, ignored, ignored, ignored, ignored, c='!') 134 135 am = ArgsManager( 136 EArgs(a='hello', b='world').args_state(ExtendArgsManager.ArgsState.original), 137 ) 138 # Or: 139 # am = ArgsManager( 140 # EArgs(ignored, ignored, ignored, ignored, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.original), 141 # ) 142 am(my_func, first, second, third, fourth, c='!') 143 """ 144 class ArgsState(Enum): 145 original = 0 146 prefix = 1 147 suffix = 2 148 manager = 3 149 150 def __init__(self, *args, **kwargs): 151 self.args = args 152 self._args_state: 'ExtendArgsManager.ArgsState' = ExtendArgsManager.ArgsState.prefix 153 self.kwargs = kwargs or dict() 154 if not isinstance(self.kwargs, Dict): 155 raise RuntimeError('Wrong parameters') 156 157 def args_state(self, args_state: 'ExtendArgsManager.ArgsState') -> 'ExtendArgsManager': 158 self._args_state = args_state 159 return self 160 161 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 162 if ExtendArgsManager.ArgsState.original == self._args_state: 163 pass 164 elif ExtendArgsManager.ArgsState.prefix == self._args_state: 165 args = tuple(list(self.args) + list(args)) 166 elif ExtendArgsManager.ArgsState.suffix == self._args_state: 167 args = tuple(list(args) + list(self.args)) 168 elif ExtendArgsManager.ArgsState.manager == self._args_state: 169 args = tuple(self.args) 170 kwargs.update(self.kwargs) 171 return args, kwargs
Usage: def my_func(first, second, third, fourth, a, b, c): ...
am = ArgsManager(
EArgs(first, second, a='hello', b='world'),
)
am(my_func, third, fourth, c='!')
am = ArgsManager(
EArgs(first, second, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.prefix),
)
am(my_func, third, fourth, c='!')
am = ArgsManager(
EArgs(third, fourth, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.suffix),
)
am(my_func, first, second, c='!')
am = ArgsManager(
EArgs(first, second, third, fourth, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.manager),
)
am(my_func, c='!')
# Or:
# am(my_func, ignored, ignored, ignored, ignored, c='!')
am = ArgsManager(
EArgs(a='hello', b='world').args_state(ExtendArgsManager.ArgsState.original),
)
# Or:
# am = ArgsManager(
# EArgs(ignored, ignored, ignored, ignored, a='hello', b='world').args_state(ExtendArgsManager.ArgsState.original),
# )
am(my_func, first, second, third, fourth, c='!')
An enumeration.
Inherited Members
- enum.Enum
- name
- value
177class ArgsManager(ArgsManagerMixin): 178 """ 179 Usage: 180 am = ArgsManager( 181 EKwargs({'a': 'hello', 'next': 'world}), 182 EKwargs(a='hello', b='world') 183 ) 184 185 Example: 186 class Item(Enum): 187 Div = 0 188 Button = 1 189 190 def html(item, color, size, step=None, length=None, strength=None) -> Any: 191 ... 192 193 am = ArgsManager( 194 EKwargs({'size': 12, 'step': 'one'}), 195 EKwargs(color='green', strength=24) 196 ) 197 198 page = list() 199 page.append(am(html, Item.Div)) 200 page.append(am(html, Item.Button)) 201 # The same as: 202 # page = list() 203 # page.append(html(Item.Div, 'green', 12, 'one', strength=24)) 204 # page.append(html(Item.Button, 'green', 12, 'one', strength=24)) 205 """ 206 def __init__(self, *args): 207 self.managers = list(args) 208 self.one_shot_managers = list() 209 self.interceptors = list() 210 self.one_shot_interceptors = list() 211 212 def append(self, manager) -> 'ArgsManager': 213 self.managers.append(manager) 214 return self 215 216 def append_one_shot(self, manager) -> 'ArgsManager': 217 self.one_shot_managers.append(manager) 218 return self 219 220 def add_interceptor(self, interceptor: Callable) -> 'ArgsManager': 221 self.interceptors.append(interceptor) 222 return self 223 224 def add_interceptor_one_shot(self, interceptor: Callable) -> 'ArgsManager': 225 self.one_shot_interceptors.append(interceptor) 226 return self 227 228 def callable(self, entity: EntityWithExtendableArgs) -> Callable: 229 def callable_entity(*args, **kwargs) -> Any: 230 original_args: Tuple[Tuple, Dict] = (copy.copy(args), copy.copy(kwargs)) 231 for manager in self.managers: 232 args, kwargs = manager(entity, *args, **kwargs) 233 234 one_shot_managers_buff = self.one_shot_managers 235 self.one_shot_managers = type(one_shot_managers_buff)() 236 for manager in one_shot_managers_buff: 237 args, kwargs = manager(entity, *args, **kwargs) 238 239 resulting_args: Tuple[Tuple, Dict] = (args, kwargs) 240 instance = entity(*args, **kwargs) 241 242 for interceptor in self.interceptors: 243 interceptor(instance, entity, original_args, resulting_args) 244 245 one_shot_interceptors_buff = self.one_shot_interceptors 246 self.one_shot_interceptors = type(one_shot_interceptors_buff)() 247 for interceptor in one_shot_interceptors_buff: 248 interceptor(instance, entity, original_args, resulting_args) 249 250 return instance 251 return callable_entity 252 253 def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any: 254 original_args: Tuple[Tuple, Dict] = (copy.copy(args), copy.copy(kwargs)) 255 for manager in self.managers: 256 args, kwargs = manager(entity, *args, **kwargs) 257 258 one_shot_managers_buff = self.one_shot_managers 259 self.one_shot_managers = type(one_shot_managers_buff)() 260 for manager in one_shot_managers_buff: 261 args, kwargs = manager(entity, *args, **kwargs) 262 263 resulting_args: Tuple[Tuple, Dict] = (args, kwargs) 264 instance = entity(*args, **kwargs) 265 for interceptor in self.interceptors: 266 interceptor(instance, entity, original_args, resulting_args) 267 268 return instance
Usage: am = ArgsManager( EKwargs({'a': 'hello', 'next': 'world}), EKwargs(a='hello', b='world') )
Example: class Item(Enum): Div = 0 Button = 1
def html(item, color, size, step=None, length=None, strength=None) -> Any:
...
am = ArgsManager(
EKwargs({'size': 12, 'step': 'one'}),
EKwargs(color='green', strength=24)
)
page = list()
page.append(am(html, Item.Div))
page.append(am(html, Item.Button))
# The same as:
# page = list()
# page.append(html(Item.Div, 'green', 12, 'one', strength=24))
# page.append(html(Item.Button, 'green', 12, 'one', strength=24))
228 def callable(self, entity: EntityWithExtendableArgs) -> Callable: 229 def callable_entity(*args, **kwargs) -> Any: 230 original_args: Tuple[Tuple, Dict] = (copy.copy(args), copy.copy(kwargs)) 231 for manager in self.managers: 232 args, kwargs = manager(entity, *args, **kwargs) 233 234 one_shot_managers_buff = self.one_shot_managers 235 self.one_shot_managers = type(one_shot_managers_buff)() 236 for manager in one_shot_managers_buff: 237 args, kwargs = manager(entity, *args, **kwargs) 238 239 resulting_args: Tuple[Tuple, Dict] = (args, kwargs) 240 instance = entity(*args, **kwargs) 241 242 for interceptor in self.interceptors: 243 interceptor(instance, entity, original_args, resulting_args) 244 245 one_shot_interceptors_buff = self.one_shot_interceptors 246 self.one_shot_interceptors = type(one_shot_interceptors_buff)() 247 for interceptor in one_shot_interceptors_buff: 248 interceptor(instance, entity, original_args, resulting_args) 249 250 return instance 251 return callable_entity
271def merge_func_args(func_list: Sequence[Callable]) -> Tuple: 272 args: List = list() 273 default_args: List = list() 274 for func in func_list: 275 is_method = None 276 if inspect.ismethod(func): 277 is_method = True 278 elif callable(func): 279 is_method = False 280 else: 281 raise RuntimeError(f'Is not callable: {repr(func)}') 282 varnames = func.__code__.co_varnames 283 if is_method: 284 if len(varnames) > 0: 285 if 'self' == varnames[0]: 286 varnames = varnames[1:] 287 spec = inspect.getfullargspec(func) 288 # print(f'ArgSpec:') 289 # pprint(spec) 290 291 if spec.defaults is None: 292 args.extend(varnames) 293 else: 294 defaults_len = len(spec.defaults) 295 args.extend(varnames[:-defaults_len]) 296 default_args.extend(varnames[-defaults_len:]) 297 # print(f'Args:') 298 # pprint(args) 299 # print(f'Default args:') 300 # pprint(default_args) 301 return args + default_args
314def func_args_to_kwargs(func, args, kwargs): 315 kwargs = copy.copy(kwargs) 316 interested_names = func.__code__.co_varnames 317 is_method = None 318 if inspect.ismethod(func): 319 is_method = True 320 elif callable(func): 321 is_method = False 322 else: 323 raise RuntimeError(f'Is not callable: {repr(func)}') 324 325 kwargs.update(zip(interested_names, args)) 326 not_needed_names = set(kwargs) - set(interested_names) 327 for name in not_needed_names: 328 kwargs.pop(name, None) 329 if is_method: 330 kwargs.pop('self', None) 331 return kwargs
342def args_kwargs_to_str(args, kwargs) -> str: 343 if args: 344 args_str = ', '.join([f'{arg}' for arg in args]) 345 else: 346 args_str = str() 347 348 if kwargs: 349 kwargs_str = ', '.join([f'{key}={value}' for key, value in kwargs.items()]) 350 else: 351 kwargs_str = str() 352 353 if kwargs_str: 354 return f'{args_str}, {kwargs_str}' 355 else: 356 return args_str
359class ArgsKwargs: 360 def __init__(self, *args, **kwargs) -> None: 361 self.args: Tuple = args 362 self.kwargs: Dict = kwargs 363 364 def __call__(self) -> Any: 365 return self.args, self.kwargs
371def prepare_arguments_positions(positional: Sequence[str], keyword_only: Sequence[str]) -> Dict[str, Optional[int]]: 372 positions: Dict[str, Optional[int]] = dict() 373 for index, name in enumerate(positional): 374 positions[name] = index 375 376 for index, name in enumerate(keyword_only): 377 positions[name] = None 378 379 return positions
Common base class for all non-exit exceptions.
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- args
386def find_arg_position_and_value(arg_name: str, positions: Dict[str, Optional[int]], args: Tuple, kwargs: Dict) -> Tuple[bool, Optional[int], Any]: 387 388 """ 389 Example: 390 from cengal.introspection.inspect import func_param_names, CodeParamNames 391 from cengal.code_flow_control.args_manager import prepare_arguments_positions, find_arg_position_and_value, UnknownArgumentError 392 393 def wrapper(arg_name: str = 'b', *args, **kwargs): 394 def my_func(a, b, *, c, d): 395 ... 396 397 params: CodeParamNames = func_param_names(my_func) 398 positoins: Dict[str, Optional[int]] = prepare_arguments_positions(params.positional, params.keyword_only) 399 found, pos, value = find_arg_position_and_value(arg_name, positoins) 400 if found: 401 print(f'Value of <{arg_name}> : {value}') 402 403 if pos is not None: 404 print(f'<{arg_name}> found as a positional argument at position {pos}') 405 406 return my_func(*args, **kwargs) 407 408 wrapper('a') 409 wrapper('a', 1, 2, 3, 4) 410 wrapper('d') 411 wrapper('d', 1, 2, 3, 4) 412 try: 413 wrapper('asdf') 414 except UnknownArgumentError as ex: 415 print('<{ex.args[1]}> is not a valid argument for my_func(). Valid arguments are: {ex.args[2]}') 416 raise 417 418 Args: 419 arg_name (str): _description_ 420 positions (Dict[str, Optional[int]]): _description_ 421 args (Tuple): _description_ 422 kwargs (Dict): _description_ 423 424 Raises: 425 UnknownArgumentError: _description_ 426 427 Returns: 428 Tuple[bool, bool, Optional[int], Any]: _description_ 429 """ 430 found: bool = False 431 pos = None 432 value = None 433 434 original_arg_pos: Optional[int] = None 435 try: 436 original_arg_pos = positions[arg_name] 437 except KeyError: 438 valid_arguments = positions.keys() 439 raise UnknownArgumentError(f'<{arg_name}> is not in {valid_arguments}', arg_name, valid_arguments) 440 441 if original_arg_pos is None: 442 found_in_args = False 443 else: 444 if len(args) > original_arg_pos: 445 found_in_args = True 446 else: 447 found_in_args = False 448 449 if found_in_args: 450 pos = original_arg_pos 451 value = args[pos] 452 found = True 453 else: 454 try: 455 value = kwargs.get(arg_name, None) 456 found = True 457 except KeyError: 458 pass 459 460 return found, pos, value
Example: from cengal.introspection.inspect import func_param_names, CodeParamNames from cengal.code_flow_control.args_manager import prepare_arguments_positions, find_arg_position_and_value, UnknownArgumentError
def wrapper(arg_name: str = 'b', *args, **kwargs):
def my_func(a, b, *, c, d):
...
params: CodeParamNames = func_param_names(my_func)
positoins: Dict[str, Optional[int]] = prepare_arguments_positions(params.positional, params.keyword_only)
found, pos, value = find_arg_position_and_value(arg_name, positoins)
if found:
print(f'Value of <{arg_name}> : {value}')
if pos is not None:
print(f'<{arg_name}> found as a positional argument at position {pos}')
return my_func(*args, **kwargs)
wrapper('a')
wrapper('a', 1, 2, 3, 4)
wrapper('d')
wrapper('d', 1, 2, 3, 4)
try:
wrapper('asdf')
except UnknownArgumentError as ex:
print('<{ex.args[1]}> is not a valid argument for my_func(). Valid arguments are: {ex.args[2]}')
raise
Args: arg_name (str): _description_ positions (Dict[str, Optional[int]]): _description_ args (Tuple): _description_ kwargs (Dict): _description_
Raises: UnknownArgumentError: _description_
Returns: Tuple[bool, bool, Optional[int], Any]: _description_
463def try_find_arg_position_and_value(arg_name: str, positions: Dict[str, Optional[int]], args: Tuple, kwargs: Dict) -> Tuple[bool, bool, Optional[int], Any]: 464 465 """ 466 Example: 467 from cengal.introspection.inspect import func_param_names, CodeParamNames 468 from cengal.code_flow_control.args_manager import prepare_arguments_positions, try_find_arg_position_and_value 469 470 def wrapper(arg_name: str = 'b', *args, **kwargs): 471 def my_func(a, b, *, c, d): 472 ... 473 474 params: CodeParamNames = func_param_names(my_func) 475 positoins: Dict[str, Optional[int]] = find_entity_arguments_positions(params.positional, params.keyword_only) 476 valid, found, pos, value = find_arg_position_and_value(arg_name, positoins) 477 if valid: 478 if found: 479 print(f'Value of <{arg_name}> : {value}') 480 481 if pos is not None: 482 print(f'<{arg_name}> found as a positional argument at position {pos}') 483 else: 484 print('<{arg_name}> is not a valid argument for my_func(). Valid arguments are: {positoins.keys()}') 485 486 return my_func(*args, **kwargs) 487 488 wrapper('a') 489 wrapper('a', 1, 2, 3, 4) 490 wrapper('d') 491 wrapper('d', 1, 2, 3, 4) 492 wrapper('asdf') 493 494 Args: 495 arg_name (str): _description_ 496 positions (Dict[str, Optional[int]]): _description_ 497 args (Tuple): _description_ 498 kwargs (Dict): _description_ 499 500 Returns: 501 Tuple[bool, bool, Optional[int], Any]: _description_ 502 """ 503 valid: bool = False 504 found: bool = False 505 pos = None 506 value = None 507 508 original_arg_pos: Optional[int] = None 509 try: 510 original_arg_pos = positions[arg_name] 511 valid = True 512 except KeyError: 513 pass 514 515 if original_arg_pos is None: 516 found_in_args = False 517 else: 518 if len(args) > original_arg_pos: 519 found_in_args = True 520 else: 521 found_in_args = False 522 523 if found_in_args: 524 pos = original_arg_pos 525 value = args[pos] 526 found = True 527 else: 528 try: 529 value = kwargs.get(arg_name, None) 530 found = True 531 except KeyError: 532 pass 533 534 return valid, found, pos, value
Example: from cengal.introspection.inspect import func_param_names, CodeParamNames from cengal.code_flow_control.args_manager import prepare_arguments_positions, try_find_arg_position_and_value
def wrapper(arg_name: str = 'b', *args, **kwargs):
def my_func(a, b, *, c, d):
...
params: CodeParamNames = func_param_names(my_func)
positoins: Dict[str, Optional[int]] = find_entity_arguments_positions(params.positional, params.keyword_only)
valid, found, pos, value = find_arg_position_and_value(arg_name, positoins)
if valid:
if found:
print(f'Value of <{arg_name}> : {value}')
if pos is not None:
print(f'<{arg_name}> found as a positional argument at position {pos}')
else:
print('<{arg_name}> is not a valid argument for my_func(). Valid arguments are: {positoins.keys()}')
return my_func(*args, **kwargs)
wrapper('a')
wrapper('a', 1, 2, 3, 4)
wrapper('d')
wrapper('d', 1, 2, 3, 4)
wrapper('asdf')
Args: arg_name (str): _description_ positions (Dict[str, Optional[int]]): _description_ args (Tuple): _description_ kwargs (Dict): _description_
Returns: Tuple[bool, bool, Optional[int], Any]: _description_