so_magic.data.backend.engine.DataEngine.create() - Code Metrics - Inspection of "docs(phi): fix warnings of phi module and test the..." - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — mpeta ( a918a8...92227f )

by Konstantinos

created 2021-04-16 14:16 UTC

so_magic.data.backend.engine.DataEngine.create() A

↳ Parent: so_magic.data.backend.engine.EngineType.dec()

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	4
dl	0
loc	6
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 so_magic.data.command_factories import MagicCommandFactory, CommandRegistrator


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

boromir674 / so-magic

Push — mpeta ( a918a8...92227f )

so_magic.data.backend.engine.DataEngine.create() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like