so_magic.data.backend.engine.DataEngine.observations() - Code Metrics - Inspection of "Release v0.5.1" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#22)

by Konstantinos

created 2021-05-08 17:52 UTC

DataEngine.observations() A

↳ Parent: so_magic.data.backend.engine

Complexity

Conditions

Size

Total Lines	13
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	2
dl	0
loc	13
rs	9.8
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'
        else:
            _ = 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__(self, item):
        if item not in self.registry:
            raise RuntimeError(f"Key '{item}' fot found in registry: [{', '.join(str(x) for x in self.registry.keys())}]")
        return self.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 EngineType(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):
        x = super().__new__(mcs, *args, **kwargs)
        x._commands = {}
        x.retriever = None
        x.iterator = None
        x.mutator = None
        x.backend = None
        x.command = mcs.magic_decorator
        x.command_factory = MagicCommandFactory()
        x._receivers = defaultdict(lambda: x._generic_cmd_receiver, observations=x._observations_from_file_cmd_receiver)
        return x

    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(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)
            # create the datapoints object and let the datapoints factory notify its listeners (eg a datapoints manager)
            _ = cls.backend.datapoints_factory.create(kwargs.get('data_structure', 'tabular-data'),
                                                      _observations, [_ for _ in []],
                                                      cls.retriever(),
                                                      cls.iterator(),
                                                      cls.mutator(),
                                                      file_path=file_path)

        return observations, {}

    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 DataEngine(metaclass=EngineType):
    """Facility to create Data Engines."""
    subclasses = {}

    @classmethod
    def new(cls, engine_name: str) -> EngineType:
        """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:
            EngineType: the Data Engine object
        """

        @DataEngine.register_as_subclass(engine_name)
        class RuntimeDataEngine(DataEngine):
            pass

        return RuntimeDataEngine

    @classmethod
    def register_as_subclass(cls, engine_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:
            engine_type (str): the name under which to register the Data Engine
        """

        def wrapper(subclass) -> type:
            cls.subclasses[engine_type] = subclass
            setattr(cls, engine_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			else:
25			_ = decorator
26			return _ # ... 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__(self, item):
39			if item not in self.registry:
40			raise RuntimeError(f"Key '{item}' fot found in registry: [{', '.join(str(x) for x in self.registry.keys())}]")
41			return self.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 EngineType(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			x = super().__new__(mcs, args, *kwargs)
64			x._commands = {}
65			x.retriever = None
66			x.iterator = None
67			x.mutator = None
68			x.backend = None
69			x.command = mcs.magic_decorator
70			x.command_factory = MagicCommandFactory()
71			x._receivers = defaultdict(lambda: x._generic_cmd_receiver, observations=x._observations_from_file_cmd_receiver)
72			return x
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(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			# create the datapoints object and let the datapoints factory notify its listeners (eg a datapoints manager)
105			_ = cls.backend.datapoints_factory.create(kwargs.get('data_structure', 'tabular-data'),
106			_observations, [_ for _ in []],
107			cls.retriever(),
108			cls.iterator(),
109			cls.mutator(),
110			file_path=file_path)
111
112			return observations, {}
113
114			def _generic_cmd_receiver(cls, callable_function, **kwargs) -> Tuple[callable, dict]:
115			"""Create the Receiver of a generic command.
116
117			It also creates the kwargs that a Command factory method would need along with the receiver object.
118
119			It is assumed that the business logic is executed in the callable function.
120
121			Args:
122			callable_function (Callable): the business logic that shall run in the command
123
124			Returns:
125			Union[callable, dict]: the receiver object that can be used to create a Command instance
126			and parameters to pass in the kwargs of the command factory
127			(eg cls.command_factory(a_function, **kwargs_dict))
128			"""
129
130			def a_function(args, *runtime_kwargs):
131			"""Just execute the business logic that is provided at runtime.
132
133			The signature of this function determines the signature that is used at runtime
134			when the command will be executed. Thus the command's arguments at runtime
135			should follow the signature of this function. So, the runtime function
136			can have any signature (since a_function uses flexible args and *runtime_kwargs).
137			"""
138			callable_function(args, *runtime_kwargs)
139
140			return a_function, {'name': lambda name: name}
141
142			def _build_command(cls, a_callable: callable, registered_name: str, data_structure='tabular-data'):
143			"""Build a command given a callable object with the business logic and register the command under a name.
144
145			Creates the required command Receiver and arguments, given a function at runtime. If the function is named
146			'observations' then the Receiver is tailored to facilitate creating a Datapoints instance given a file path
147			with the raw data.
148
149			Args:
150			a_callable (Callable): holds the business logic that executes when the command shall be executed
151			registered_name (str): the name under which to register the command (can be used to reference the command)
152			data_structure (str, optional): useful when creating a command that instantiates Datapoints objects.
153			Defaults to 'tabular-data'.
154			"""
155			receiver, kwargs_data = cls._receivers[registered_name](a_callable, data_structure=data_structure)
156			cls.registry[registered_name] = receiver
157			cls._commands[registered_name] = cls.command_factory(receiver, {k: v for k, v in dict(kwargs_data, {
158			'name': kwargs_data.get('name', lambda name: '')(registered_name)}).items() if v})
159
160			def dec(cls, data_structure='tabular-data') -> Callable[[Callable], Callable]:
161			"""Register a new command that executes the business logic supplied at runtime.
162
163			Decorate a function so that its body acts as the business logic that runs as part of a Command.
164			The name of the function can be used to later reference the Command (or a prototype object of the Command).
165
166			Using the 'observations' name for your function will register a command that upon execution creates a new
167			instance of Datapoints (see Datapoints class), provided that the runtime function returns an object that acts as
168			the 'observations' attribute of a Datapoints object.
169
170			Args:
171			data_structure (str, optional): useful when the function name is 'observations'. Defaults to 'tabular-data'.
172			"""
173
174			def wrapper(a_callable: Callable) -> Callable:
175			"""Build and register a new Command given a callable object that holds the important business logic.
176
177			Args:
178			a_callable (Callable): the Command's important underlying business logic
179			"""
180			if hasattr(a_callable, '__code__'): # a_callable object has been defined with the def python keyword
181			decorated_function_name = a_callable.__code__.co_name
182			cls._build_command(a_callable, decorated_function_name, data_structure=data_structure)
183			else:
184			raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
185			return a_callable
186
187			return wrapper
188
189
190			class DataEngine(metaclass=EngineType):
191			"""Facility to create Data Engines."""
192			subclasses = {}
193
194			@classmethod
195			def new(cls, engine_name: str) -> EngineType:
196			"""Create a Data Engine object and register it under the given name, to be able to reference it by name.
197
198			Creates a Data Engine that serves as an empty canvas to add attributes and Commands.
199
200			Args:
201			engine_name (str): the name under which to register the Data Engine
202
203			Returns:
204			EngineType: the Data Engine object
205			"""
206
207			@DataEngine.register_as_subclass(engine_name)
208			class RuntimeDataEngine(DataEngine):
209			pass
210
211			return RuntimeDataEngine
212
213			@classmethod
214			def register_as_subclass(cls, engine_type: str):
215			"""Indicate that a class is a subclass of DataEngine and register it under the given name.
216
217			It also sets the engine_type attribute on the decorate class to be equal to the subclass.
218
219			Args:
220			engine_type (str): the name under which to register the Data Engine
221			"""
222
223			def wrapper(subclass) -> type:
224			cls.subclasses[engine_type] = subclass
225			setattr(cls, engine_type, subclass)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-08-09 18:51 UTC by Report Bug Copy Issue Report The variable `engine_type` does not seem to be defined. Loading history...
226			return subclass
227
228			return wrapper
229

boromir674 / so-magic

Pull Request — master (#22)

DataEngine.observations() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like