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
EntityWithExtendableArgs = typing.Union[typing.Type, typing.Callable]
class ArgsManagerMixin:
49class ArgsManagerMixin:
50    def __call__(self, entity: EntityWithExtendableArgs, *args, **kwargs) -> Any:
51        raise NotImplementedError
class EntityArgsHolder:
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
EntityArgsHolder(entity: typing.Union[typing.Type, typing.Callable], *args, **kwargs)
55    def __init__(self, entity: EntityWithExtendableArgs, *args, **kwargs):
56        self.entity: EntityWithExtendableArgs = entity
57        self.args: Tuple = args
58        self.kwargs: Dict = kwargs
entity: Union[Type, Callable]
args: Tuple
kwargs: Dict
def args_kwargs(self) -> Tuple[Tuple, Dict]:
63    def args_kwargs(self) -> Tuple[Tuple, Dict]:
64        return self.args, self.kwargs
def entity_args_kwargs(self) -> Tuple[Union[Type, Callable], Tuple, Dict]:
66    def entity_args_kwargs(self) -> Tuple[EntityWithExtendableArgs, Tuple, Dict]:
67        return self.entity, self.args, self.kwargs
EAH = <class 'EntityArgsHolder'>
class EntityArgsHolderExplicit(EntityArgsHolder):
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
EntityArgsHolderExplicit(entity: typing.Union[typing.Type, typing.Callable], args, kwargs)
74    def __init__(self, entity: EntityWithExtendableArgs, args, kwargs):
75        super().__init__(entity)
76        self.args: Tuple = args
77        self.kwargs: Dict = kwargs
args: Tuple
kwargs: Dict
EAHE = <class 'EntityArgsHolderExplicit'>
class ExtendKwargsManager(ArgsManagerMixin):
 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') )

ExtendKwargsManager(*args, **kwargs)
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')
kwargs
EKwargs = <class 'ExtendKwargsManager'>
class ExtendArgsManager(ArgsManagerMixin):
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='!')
ExtendArgsManager(*args, **kwargs)
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')
args
kwargs
def args_state( self, args_state: ExtendArgsManager.ArgsState) -> ExtendArgsManager:
157    def args_state(self, args_state: 'ExtendArgsManager.ArgsState') -> 'ExtendArgsManager':
158        self._args_state = args_state
159        return self
class ExtendArgsManager.ArgsState(enum.Enum):
144    class ArgsState(Enum):
145        original = 0
146        prefix = 1
147        suffix = 2
148        manager = 3

An enumeration.

original = <ArgsState.original: 0>
prefix = <ArgsState.prefix: 1>
suffix = <ArgsState.suffix: 2>
manager = <ArgsState.manager: 3>
Inherited Members
enum.Enum
name
value
EArgs = <class 'ExtendArgsManager'>
class ArgsManager(ArgsManagerMixin):
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))
ArgsManager(*args)
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()
managers
one_shot_managers
interceptors
one_shot_interceptors
def append( self, manager) -> ArgsManager:
212    def append(self, manager) -> 'ArgsManager':
213        self.managers.append(manager)
214        return self
def append_one_shot( self, manager) -> ArgsManager:
216    def append_one_shot(self, manager) -> 'ArgsManager':
217        self.one_shot_managers.append(manager)
218        return self
def add_interceptor( self, interceptor: typing.Callable) -> ArgsManager:
220    def add_interceptor(self, interceptor: Callable) -> 'ArgsManager':
221        self.interceptors.append(interceptor)
222        return self
def add_interceptor_one_shot( self, interceptor: typing.Callable) -> ArgsManager:
224    def add_interceptor_one_shot(self, interceptor: Callable) -> 'ArgsManager':
225        self.one_shot_interceptors.append(interceptor)
226        return self
def callable(self, entity: typing.Union[typing.Type, typing.Callable]) -> Callable:
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
def merge_func_args(func_list: typing.Sequence[typing.Callable]) -> Tuple:
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
def interested_args_to_kwargs(interested_args, args, kwargs):
304def interested_args_to_kwargs(interested_args, args, kwargs):
305    kwargs = copy.copy(kwargs)
306    absent_args = list()
307    for arg in interested_args:
308        if arg not in kwargs:
309            absent_args.append(arg)
310    kwargs.update(zip(absent_args, args))
311    return kwargs
def func_args_to_kwargs(func, args, kwargs):
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
def number_of_provided_args(args, kwargs):
334def number_of_provided_args(args, kwargs):
335    return len(args) + len(kwargs)
def args_kwargs(*args, **kwargs) -> Tuple[Tuple, Dict]:
338def args_kwargs(*args, **kwargs) -> Tuple[Tuple, Dict]:
339    return args, kwargs
def args_kwargs_to_str(args, kwargs) -> str:
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
class ArgsKwargs:
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
ArgsKwargs(*args, **kwargs)
360    def __init__(self, *args, **kwargs) -> None:
361        self.args: Tuple = args
362        self.kwargs: Dict = kwargs
args: Tuple
kwargs: Dict
AK = <class 'ArgsKwargs'>
def prepare_arguments_positions( positional: typing.Sequence[str], keyword_only: typing.Sequence[str]) -> Dict[str, Union[int, NoneType]]:
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
class UnknownArgumentError(builtins.Exception):
382class UnknownArgumentError(Exception):
383    pass

Common base class for all non-exit exceptions.

Inherited Members
builtins.Exception
Exception
builtins.BaseException
with_traceback
args
def find_arg_position_and_value( arg_name: str, positions: typing.Dict[str, typing.Union[int, NoneType]], args: typing.Tuple, kwargs: typing.Dict) -> Tuple[bool, Union[int, NoneType], Any]:
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_

def try_find_arg_position_and_value( arg_name: str, positions: typing.Dict[str, typing.Union[int, NoneType]], args: typing.Tuple, kwargs: typing.Dict) -> Tuple[bool, bool, Union[int, NoneType], Any]:
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_