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

Passed

Push — datapoints-package ( a11eff )

by Konstantinos

created 2021-03-26 14:41 UTC

A

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

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

from collections import defaultdict
from typing import Tuple
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, **receiver_kwargs) -> Tuple[callable, dict]:
        """Create the Receiver of a command that creates datapoints from a file.
        
        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 observations(file_path, **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
            """
            _observations = callable_function(file_path, **kwargs)
            datapoints = cls.backend.datapoints_factory.create(receiver_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, **receiver_kwargs):
        """Create the Receiver of a generic command.
        
        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, **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 **kwargs).
            
            Args:
                file_path (str): the file in disk that contains the data to be read into observations
            """
            callable_function(*args, **kwargs)       
        return a_function, {'name': lambda name: name}

    def _build_command_receiver(cls, a_callable: callable, decorated_function_name: str, data_structure='tabular-data'):
        receiver, kwargs_data = cls._receivers[decorated_function_name](a_callable, data_structure=data_structure)
        cls.registry[decorated_function_name] = receiver
        cls._commands[decorated_function_name] = cls.command_factory(receiver, **{k:v for k,v in dict(kwargs_data, **{'name': kwargs_data.get('name', lambda name: '')(decorated_function_name)}).items() if v})

    def dec(cls, data_structure='tabular-data'):
        def wrapper(a_callable):
            if hasattr(a_callable, '__code__'):  # it a function (def func_name ..)
                name = a_callable.__code__.co_name
                cls._build_command_receiver(a_callable, 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):
    subclasses = {}

    @classmethod
    def new(cls, engine_name):
        @DataEngine.register_as_subclass(engine_name)
        class RuntimeDataEngine(DataEngine):
            pass
        return RuntimeDataEngine

    @classmethod
    def register_as_subclass(cls, engine_type):
        def wrapper(subclass):
            cls.subclasses[engine_type] = subclass
            setattr(cls, engine_type, subclass)

            return subclass
        return wrapper

    @classmethod
    def create(cls, engine_type, *args, **kwargs):
        if engine_type not in cls.subclasses:
            raise ValueError(
                f"Request Engine of type '{engine_type}'; supported are [{', '.join(sorted(cls.subclasses.keys()))}]")
        return cls.subclasses[engine_type](*args, **kwargs)


1			from collections import defaultdict
2			from typing import Tuple
3			from so_magic.data.command_factories import MagicCommandFactory, CommandRegistrator
4
5
6			class EngineType(CommandRegistrator):
7			"""Tabular Data Backend type representation.
8
9			Classes using this class as metaclass gain certain class attributes such as
10			attributes related to tabular data operations (retriever, iterator, mutator)
11			and attributes related to constructing command object prototypes
12			(command_factory attribute).
13			"""
14			def __new__(mcs, args, *kwargs):
15			x = super().__new__(mcs, args, *kwargs)
16			x._commands = {}
17			x.retriever = None
18			x.iterator = None
19			x.mutator = None
20			x.backend = None
21			x.command = mcs.magic_decorator
22			x.command_factory = MagicCommandFactory()
23			x._receivers = defaultdict(lambda: x._generic_cmd_receiver, observations=x._observations_from_file_cmd_receiver)
24			return x
25
26			def _observations_from_file_cmd_receiver(cls, callable_function, **receiver_kwargs) -> Tuple[callable, dict]:
27			"""Create the Receiver of a command that creates datapoints from a file.
28
29			It is assumed that the business logic is executed in the callable
30			function.
31
32			Args:
33			callable_function (callable): the business logic that shall run in the command
34
35			Returns:
36			Union[callable, dict]: the receiver object that can be used to create a Command instance
37			and parameters to pass in the kwargs of the command factory (eg cls.command_factory(a_function, **kwargs_dict))
38			"""
39			def observations(file_path, **kwargs):
40			"""Construct the observations attribute of a Datapoints instance.
41
42			The signature of this function determines the signature that is used at runtime
43			when the command will be executed. Thus the command's arguments at runtime
44			should follow the signature of this function.
45
46			Args:
47			file_path (str): the file in disk that contains the data to be read into observations
48			"""
49			_observations = callable_function(file_path, **kwargs)
50			datapoints = cls.backend.datapoints_factory.create(receiver_kwargs.get('data_structure', 'tabular-data'), _observations, [_ for _ in []],
51			cls.retriever(),
52			cls.iterator(),
53			cls.mutator(),
54			file_path=file_path)
55			return observations, {}
56
57			def _generic_cmd_receiver(cls, callable_function, **receiver_kwargs):
58			"""Create the Receiver of a generic command.
59
60			It is assumed that the business logic is executed in the callable
61			function.
62
63			Args:
64			callable_function (callable): the business logic that shall run in the command
65
66			Returns:
67			Union[callable, dict]: the receiver object that can be used to create a Command instance
68			and parameters to pass in the kwargs of the command factory (eg cls.command_factory(a_function, **kwargs_dict))
69			"""
70			def a_function(args, *kwargs):
71			"""Just execute the business logic that is provided at runtime.
72
73			The signature of this function determines the signature that is used at runtime
74			when the command will be executed. Thus the command's arguments at runtime
75			should follow the signature of this function. So, the runtime function
76			can have any signature (since a_function uses flexible args and *kwargs).
77
78			Args:
79			file_path (str): the file in disk that contains the data to be read into observations
80			"""
81			callable_function(args, *kwargs)
82			return a_function, {'name': lambda name: name}
83
84			def _build_command_receiver(cls, a_callable: callable, decorated_function_name: str, data_structure='tabular-data'):
85			receiver, kwargs_data = cls._receivers[decorated_function_name](a_callable, data_structure=data_structure)
86			cls.registry[decorated_function_name] = receiver
87			cls._commands[decorated_function_name] = cls.command_factory(receiver, {k:v for k,v in dict(kwargs_data, {'name': kwargs_data.get('name', lambda name: '')(decorated_function_name)}).items() if v})
88
89			def dec(cls, data_structure='tabular-data'):
90			def wrapper(a_callable):
91			if hasattr(a_callable, '__code__'): # it a function (def func_name ..)
92			name = a_callable.__code__.co_name
93			cls._build_command_receiver(a_callable, name, data_structure=data_structure)
94			else:
95			raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
96			return a_callable
97			return wrapper
98
99
100			class DataEngine(metaclass=EngineType):
101			subclasses = {}
102
103			@classmethod
104			def new(cls, engine_name):
105			@DataEngine.register_as_subclass(engine_name)
106			class RuntimeDataEngine(DataEngine):
107			pass
108			return RuntimeDataEngine
109
110			@classmethod
111			def register_as_subclass(cls, engine_type):
112			def wrapper(subclass):
113			cls.subclasses[engine_type] = subclass
114			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...
115			return subclass
116			return wrapper
117
118			@classmethod
119			def create(cls, engine_type, args, *kwargs):
120			if engine_type not in cls.subclasses:
121			raise ValueError(
122			f"Request Engine of type '{engine_type}'; supported are [{', '.join(sorted(cls.subclasses.keys()))}]")
123			return cls.subclasses[engine_type](args, *kwargs)
124

boromir674 / so-magic

Push — datapoints-package ( a11eff )

A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like