so_magic.data.backend.backend.CommandRegistrator.__new__() - Code Metrics - Inspection of "Automate Operations" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#32)

by Konstantinos

created 2021-06-22 14:20 UTC

CommandRegistrator.new() A

↳ Parent: so_magic.data.backend.backend

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	3
dl	0
loc	5
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 magic_decorator(mcs, arg=None):
        def decorator(_func):
            def wrapper(*a, **ka):
                ffunc = a[0]
                mcs._wrapper(ffunc, *a[1:], **ka)
                return ffunc
            return wrapper

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

boromir674 / so-magic

Pull Request — dev (#32)

CommandRegistrator.__new__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

CommandRegistrator.new() A