so_magic.data.backend.backend.MyDecorator.wrapper() - Code Metrics - Inspection of "feat(magic-decorator): implement a type class that..." - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — mpeta ( 9fae00...4519ef )

by Konstantinos

created 2021-06-27 18:13 UTC

MyDecorator.wrapper() A

↳ Parent: so_magic.data.backend.backend

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	4
dl	0
loc	4
rs	10
c	0
b	0
f	0

"""This module defines a way to create Data Engines and to register new commands that a Data Engine can execute.
"""
from collections import defaultdict
from typing import Tuple, Callable
from .engine_command_factory import MagicCommandFactory


class MyDecorator(type):
    """Metaclass that provides a decorator able to be invoked both with and without parenthesis.
    The wrapper function logic should be implemented by the client code.
    """
    @classmethod
    def wrapper(cls, func, *args, **kwargs):
        result = func(*args, **kwargs)
        return result
    @classmethod
    def magic_decorator(cls, arg=None):
        def decorator(func):
            def wrapper(*a, **ka):
                return cls.wrapper(func, *a, **ka)
            return wrapper

        if callable(arg):
            return decorator(arg)  # return 'wrapper'
        else:
            return decorator  # ... or 'decorator'


class CommandRegistrator(MyDecorator):
    """Classes can use this class as metaclass to obtain a single registration point accessible as class attribute.
    """
    def __new__(mcs, *args, **kwargs):
        class_object = super().__new__(mcs, *args, **kwargs)
        class_object.state = None
        class_object.registry = {}
        return class_object

    def __getitem__(cls, item):
        if item not in cls.registry:
            raise RuntimeError(f"Key '{item}' fot found in registry: "
                               f"[{', '.join(str(x) for x in cls.registry.keys())}]")
        return cls.registry[item]

    # Legacy feature, not currently used in production
    def func_decorator(cls):
        def wrapper(a_callable):
            if hasattr(a_callable, '__code__'):  # it a function (def func_name ..)
                cls.registry[a_callable.__code__.co_name] = a_callable
            else:
                raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
            return a_callable
        return wrapper


class BackendType(CommandRegistrator):
    """Tabular Data Backend type representation.

    Classes using this class as metaclass gain certain class attributes such as
    attributes related to tabular data operations (retriever, iterator, mutator) and attributes related to constructing
    command object prototypes (command_factory attribute).
    """

    def __new__(mcs, *args, **kwargs):
        engine_type = super().__new__(mcs, *args, **kwargs)
        engine_type._commands = {}
        engine_type.retriever = None
        engine_type.iterator = None
        engine_type.mutator = None
        engine_type.datapoints_factory = None
        engine_type.command_factory = MagicCommandFactory()
        engine_type._receivers = defaultdict(lambda: engine_type._generic_cmd_receiver,
                                             observations_command=engine_type._observations_from_file_cmd_receiver)
        return engine_type

    def _observations_from_file_cmd_receiver(cls, callable_function, **kwargs) -> Tuple[callable, dict]:
        """Create the Receiver of a command that creates datapoints from a file.

        It also creates the kwargs that a Command factory method would need along with the receiver object.

        It is assumed that the business logic is executed in the callable function supplied.
        You can use the data_structure "keyword" argument (kwarg) to indicate how should we parse/read
        the raw data from the file. Supported values: 'tabular-data'

        Args:
            callable_function (callable): the business logic that shall run in the command

        Returns:
            Union[callable, dict]: the receiver object that can be used to create a Command instance
                                    and parameters to pass in the kwargs of the command factory method (eg
                                    cls.command_factory(a_function, **kwargs_dict))
        """

        def observations_command(file_path, **runtime_kwargs):
            """Construct the observations attribute of a Datapoints instance.

            The signature of this function determines the signature that is used at runtime
            when the command will be executed. Thus the command's arguments at runtime
            should follow the signature of this function.

            Args:
                file_path (str): the file in disk that contains the data to be read into observations
            """
            # create the observations object
            _observations = callable_function(file_path, **runtime_kwargs)
            _ = cls.datapoints_factory.create(kwargs.get('data_structure', 'tabular-data'),
                                              _observations, [],
                                              cls.retriever(),
                                              cls.iterator(),
                                              cls.mutator(),
                                              file_path=file_path)
        return observations_command, {}

    def _generic_cmd_receiver(cls, callable_function, **kwargs) -> Tuple[callable, dict]:
        """Create the Receiver of a generic command.

        It also creates the kwargs that a Command factory method would need along with the receiver object.

        It is assumed that the business logic is executed in the callable function.

        Args:
            callable_function (Callable): the business logic that shall run in the command

        Returns:
            Union[callable, dict]: the receiver object that can be used to create a Command instance
                                    and parameters to pass in the kwargs of the command factory
                                    (eg cls.command_factory(a_function, **kwargs_dict))
        """

        def a_function(*args, **runtime_kwargs):
            """Just execute the business logic that is provided at runtime.

            The signature of this function determines the signature that is used at runtime
            when the command will be executed. Thus the command's arguments at runtime
            should follow the signature of this function. So, the runtime function
            can have any signature (since a_function uses flexible *args and **runtime_kwargs).
            """
            callable_function(*args, **runtime_kwargs)

        return a_function, {'name': lambda name: name}

    def _build_command(cls, a_callable: callable, registered_name: str, data_structure='tabular-data'):
        """Build a command given a callable object with the business logic and register the command under a name.

        Creates the required command Receiver and arguments, given a function at runtime. If the function is named
        'observations' then the Receiver is tailored to facilitate creating a Datapoints instance given a file path
        with the raw data.

        Args:
            a_callable (Callable): holds the business logic that executes when the command shall be executed
            registered_name (str): the name under which to register the command (can be used to reference the command)
            data_structure (str, optional): useful when creating a command that instantiates Datapoints objects.
            Defaults to 'tabular-data'.
        """
        receiver, kwargs_data = cls._receivers[registered_name](a_callable, data_structure=data_structure)
        cls.registry[registered_name] = receiver
        cls._commands[registered_name] = cls.command_factory(receiver, **{k: v for k, v in dict(kwargs_data, **{
            'name': kwargs_data.get('name', lambda name: '')(registered_name)}).items() if v})

    def dec(cls, data_structure='tabular-data') -> Callable[[Callable], Callable]:
        """Register a new command that executes the business logic supplied at runtime.

        Decorate a function so that its body acts as the business logic that runs as part of a Command.
        The name of the function can be used to later reference the Command (or a prototype object of the Command).

        Using the 'observations' name for your function will register a command that upon execution creates a new
        instance of Datapoints (see Datapoints class), provided that the runtime function returns an object that acts as
        the 'observations' attribute of a Datapoints object.

        Args:
            data_structure (str, optional): useful when the function name is 'observations'. Defaults to 'tabular-data'.
        """

        def wrapper(a_callable: Callable) -> Callable:
            """Build and register a new Command given a callable object that holds the important business logic.

            Args:
                a_callable (Callable): the Command's important underlying business logic
            """
            if hasattr(a_callable, '__code__'):  # a_callable object has been defined with the def python keyword
                decorated_function_name = a_callable.__code__.co_name
                cls._build_command(a_callable, decorated_function_name, data_structure=data_structure)
            else:
                raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
            return a_callable

        return wrapper


class EngineBackend(metaclass=BackendType):
    """Facility to create Data Engines."""
    subclasses = {}

    @classmethod
    def new(cls, engine_name: str) -> BackendType:
        """Create a Data Engine object and register it under the given name, to be able to reference it by name.

        Creates a Data Engine that serves as an empty canvas to add attributes and Commands.

        Args:
            engine_name (str): the name under which to register the Data Engine

        Returns:
            BackendType: the Data Engine object
        """

        @EngineBackend.register_as_subclass(engine_name)
        class RuntimeEngineBackend(EngineBackend): pass

        return RuntimeEngineBackend

    @classmethod
    def register_as_subclass(cls, backend_type: str):
        """Indicate that a class is a subclass of DataEngine and register it under the given name.

        It also sets the engine_type attribute on the decorate class to be equal to the subclass.

        Args:
            backend_type (str): the name under which to register the Data Engine
        """

        def wrapper(subclass) -> type:
            cls.subclasses[backend_type] = subclass
            setattr(cls, backend_type, subclass)

            return subclass

        return wrapper


1			"""This module defines a way to create Data Engines and to register new commands that a Data Engine can execute.
2			"""
3			from collections import defaultdict
4			from typing import Tuple, Callable
5			from .engine_command_factory import MagicCommandFactory
6
7
8			class MyDecorator(type):
9			"""Metaclass that provides a decorator able to be invoked both with and without parenthesis.
10			The wrapper function logic should be implemented by the client code.
11			"""
12			@classmethod
13			def wrapper(cls, func, args, *kwargs):
14			result = func(args, *kwargs)
15			return result
16			@classmethod
17			def magic_decorator(cls, arg=None):
18			def decorator(func):
19			def wrapper(a, *ka):
20			return cls.wrapper(func, a, *ka)
21			return wrapper
22
23			if callable(arg):
24			return decorator(arg) # return 'wrapper'
25			else:
26			return decorator # ... or 'decorator'
27
28
29			class CommandRegistrator(MyDecorator):
30			"""Classes can use this class as metaclass to obtain a single registration point accessible as class attribute.
31			"""
32			def __new__(mcs, args, *kwargs):
33			class_object = super().__new__(mcs, args, *kwargs)
34			class_object.state = None
35			class_object.registry = {}
36			return class_object
37
38			def __getitem__(cls, item):
39			if item not in cls.registry:
40			raise RuntimeError(f"Key '{item}' fot found in registry: "
41			f"[{', '.join(str(x) for x in cls.registry.keys())}]")
42			return cls.registry[item]
43
44			# Legacy feature, not currently used in production
45			def func_decorator(cls):
46			def wrapper(a_callable):
47			if hasattr(a_callable, '__code__'): # it a function (def func_name ..)
48			cls.registry[a_callable.__code__.co_name] = a_callable
49			else:
50			raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
51			return a_callable
52			return wrapper
53
54
55			class BackendType(CommandRegistrator):
56			"""Tabular Data Backend type representation.
57
58			Classes using this class as metaclass gain certain class attributes such as
59			attributes related to tabular data operations (retriever, iterator, mutator) and attributes related to constructing
60			command object prototypes (command_factory attribute).
61			"""
62
63			def __new__(mcs, args, *kwargs):
64			engine_type = super().__new__(mcs, args, *kwargs)
65			engine_type._commands = {}
66			engine_type.retriever = None
67			engine_type.iterator = None
68			engine_type.mutator = None
69			engine_type.datapoints_factory = None
70			engine_type.command_factory = MagicCommandFactory()
71			engine_type._receivers = defaultdict(lambda: engine_type._generic_cmd_receiver,
72			observations_command=engine_type._observations_from_file_cmd_receiver)
73			return engine_type
74
75			def _observations_from_file_cmd_receiver(cls, callable_function, **kwargs) -> Tuple[callable, dict]:
76			"""Create the Receiver of a command that creates datapoints from a file.
77
78			It also creates the kwargs that a Command factory method would need along with the receiver object.
79
80			It is assumed that the business logic is executed in the callable function supplied.
81			You can use the data_structure "keyword" argument (kwarg) to indicate how should we parse/read
82			the raw data from the file. Supported values: 'tabular-data'
83
84			Args:
85			callable_function (callable): the business logic that shall run in the command
86
87			Returns:
88			Union[callable, dict]: the receiver object that can be used to create a Command instance
89			and parameters to pass in the kwargs of the command factory method (eg
90			cls.command_factory(a_function, **kwargs_dict))
91			"""
92
93			def observations_command(file_path, **runtime_kwargs):
94			"""Construct the observations attribute of a Datapoints instance.
95
96			The signature of this function determines the signature that is used at runtime
97			when the command will be executed. Thus the command's arguments at runtime
98			should follow the signature of this function.
99
100			Args:
101			file_path (str): the file in disk that contains the data to be read into observations
102			"""
103			# create the observations object
104			_observations = callable_function(file_path, **runtime_kwargs)
105			_ = cls.datapoints_factory.create(kwargs.get('data_structure', 'tabular-data'),
106			_observations, [],
107			cls.retriever(),
108			cls.iterator(),
109			cls.mutator(),
110			file_path=file_path)
111			return observations_command, {}
112
113			def _generic_cmd_receiver(cls, callable_function, **kwargs) -> Tuple[callable, dict]:
114			"""Create the Receiver of a generic command.
115
116			It also creates the kwargs that a Command factory method would need along with the receiver object.
117
118			It is assumed that the business logic is executed in the callable function.
119
120			Args:
121			callable_function (Callable): the business logic that shall run in the command
122
123			Returns:
124			Union[callable, dict]: the receiver object that can be used to create a Command instance
125			and parameters to pass in the kwargs of the command factory
126			(eg cls.command_factory(a_function, **kwargs_dict))
127			"""
128
129			def a_function(args, *runtime_kwargs):
130			"""Just execute the business logic that is provided at runtime.
131
132			The signature of this function determines the signature that is used at runtime
133			when the command will be executed. Thus the command's arguments at runtime
134			should follow the signature of this function. So, the runtime function
135			can have any signature (since a_function uses flexible args and *runtime_kwargs).
136			"""
137			callable_function(args, *runtime_kwargs)
138
139			return a_function, {'name': lambda name: name}
140
141			def _build_command(cls, a_callable: callable, registered_name: str, data_structure='tabular-data'):
142			"""Build a command given a callable object with the business logic and register the command under a name.
143
144			Creates the required command Receiver and arguments, given a function at runtime. If the function is named
145			'observations' then the Receiver is tailored to facilitate creating a Datapoints instance given a file path
146			with the raw data.
147
148			Args:
149			a_callable (Callable): holds the business logic that executes when the command shall be executed
150			registered_name (str): the name under which to register the command (can be used to reference the command)
151			data_structure (str, optional): useful when creating a command that instantiates Datapoints objects.
152			Defaults to 'tabular-data'.
153			"""
154			receiver, kwargs_data = cls._receivers[registered_name](a_callable, data_structure=data_structure)
155			cls.registry[registered_name] = receiver
156			cls._commands[registered_name] = cls.command_factory(receiver, {k: v for k, v in dict(kwargs_data, {
157			'name': kwargs_data.get('name', lambda name: '')(registered_name)}).items() if v})
158
159			def dec(cls, data_structure='tabular-data') -> Callable[[Callable], Callable]:
160			"""Register a new command that executes the business logic supplied at runtime.
161
162			Decorate a function so that its body acts as the business logic that runs as part of a Command.
163			The name of the function can be used to later reference the Command (or a prototype object of the Command).
164
165			Using the 'observations' name for your function will register a command that upon execution creates a new
166			instance of Datapoints (see Datapoints class), provided that the runtime function returns an object that acts as
167			the 'observations' attribute of a Datapoints object.
168
169			Args:
170			data_structure (str, optional): useful when the function name is 'observations'. Defaults to 'tabular-data'.
171			"""
172
173			def wrapper(a_callable: Callable) -> Callable:
174			"""Build and register a new Command given a callable object that holds the important business logic.
175
176			Args:
177			a_callable (Callable): the Command's important underlying business logic
178			"""
179			if hasattr(a_callable, '__code__'): # a_callable object has been defined with the def python keyword
180			decorated_function_name = a_callable.__code__.co_name
181			cls._build_command(a_callable, decorated_function_name, data_structure=data_structure)
182			else:
183			raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
184			return a_callable
185
186			return wrapper
187
188
189			class EngineBackend(metaclass=BackendType):
190			"""Facility to create Data Engines."""
191			subclasses = {}
192
193			@classmethod
194			def new(cls, engine_name: str) -> BackendType:
195			"""Create a Data Engine object and register it under the given name, to be able to reference it by name.
196
197			Creates a Data Engine that serves as an empty canvas to add attributes and Commands.
198
199			Args:
200			engine_name (str): the name under which to register the Data Engine
201
202			Returns:
203			BackendType: the Data Engine object
204			"""
205
206			@EngineBackend.register_as_subclass(engine_name)
207			class RuntimeEngineBackend(EngineBackend): pass
208
209			return RuntimeEngineBackend
210
211			@classmethod
212			def register_as_subclass(cls, backend_type: str):
213			"""Indicate that a class is a subclass of DataEngine and register it under the given name.
214
215			It also sets the engine_type attribute on the decorate class to be equal to the subclass.
216
217			Args:
218			backend_type (str): the name under which to register the Data Engine
219			"""
220
221			def wrapper(subclass) -> type:
222			cls.subclasses[backend_type] = subclass
223			setattr(cls, backend_type, subclass)
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-06-02 19:15 UTC by Report Bug Copy Issue Report The variable `backend_type` does not seem to be defined. Loading history...
224			return subclass
225
226			return wrapper
227

boromir674 / so-magic

Push — mpeta ( 9fae00...4519ef )

MyDecorator.wrapper() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like