Skip to content

Processor

dandy.core.processor.processor.BaseProcessor

Bases: ABC

Base class for all processing classes in dandy.

description = None class-attribute instance-attribute

process abstractmethod classmethod

This method has hooks on it to allow for easy debugging

Parameters:

  • args –

    Arguments

  • kwargs –

    Keyword Arguments

Returns:

  • Any –

    Any

Source code in dandy/core/processor/processor.py 
16
17
18
19
20
21
22
23
24
25
@classmethod
@abstractmethod
def process(cls, *args, **kwargs) -> Any:
    """
    This method has hooks on it to allow for easy debugging    
    :param args: Arguments 
    :param kwargs: Keyword Arguments
    :return: Any
    """
    raise NotImplementedError

process_to_future classmethod

This method is used to generate an AsyncFuture of the process method

Parameters:

  • args –

    Arguments

  • kwargs –

    Keyword Arguments

Returns:

Source code in dandy/core/processor/processor.py 
27
28
29
30
31
32
33
34
35
@classmethod
def process_to_future(cls, *args, **kwargs) -> AsyncFuture:
    """
    This method is used to generate an AsyncFuture of the process method
    :param args: Arguments
    :param kwargs: Keyword Arguments
    :return: AsyncFuture
    """
    return AsyncFuture(cls.process, *args, **kwargs)

dandy.core.processor.abc_meta.ProcessorABCMeta

Bases: ABCMeta

__new__

Source code in dandy/core/processor/abc_meta.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __new__(cls, name, bases, dct):
    processor_class = super().__new__(cls, name, bases, dct)

    if 'process' in dct:
        original_process = dct['process']

        if isinstance(original_process, classmethod):
            original_func = original_process.__func__
            original_process_signature = inspect.signature(original_func)

            def wrapped_process(cls, *args, **kwargs):
                if getattr(cls, "_recorder_called", None) is None:
                    cls._recorder_event_id = generate_new_recorder_event_id()

                if Recorder.is_recording and not getattr(cls, "_recorder_called", False):
                    Recorder.add_event(
                        Event(
                            id=cls._recorder_event_id,
                            object_name=pascal_to_title_case(cls.__qualname__),
                            callable_name='Process',
                            type=EventType.RUN,
                            attributes=[
                                EventAttribute(
                                    key='Module',
                                    value=cls.__module__,
                                ),
                                EventAttribute(
                                    key='Args',
                                    value=json.dumps(
                                        args,
                                        indent=4,
                                        default=json_default
                                    ),
                                    is_card=True
                                ),
                                EventAttribute(
                                    key='Kwargs',
                                    value=json.dumps(
                                        kwargs,
                                        indent=4,
                                        default=json_default
                                    ),
                                    is_card=True
                                )
                            ],
                        )
                    )

                    cls._recorder_called = True

                result = original_func(cls, *args, **kwargs)

                if Recorder.is_recording and getattr(cls, "_recorder_called", True):
                    Recorder.add_event(
                        Event(
                            id=cls._recorder_event_id,
                            object_name=pascal_to_title_case(cls.__qualname__),
                            callable_name='Process Returned Result',
                            type=EventType.RESULT,
                            attributes=[
                                EventAttribute(
                                    key='Module',
                                    value=cls.__module__,
                                ),
                                EventAttribute(
                                    key='Returned Result',
                                    value=json.dumps(
                                        result,
                                        indent=4,
                                        default=json_default
                                    ),
                                    is_card=True,
                                )
                            ],
                        )
                    )
                    cls._recorder_called = False

                return result

            wrapped_process.__signature__ = original_process_signature

            setattr(processor_class, 'process', classmethod(wrapped_process))

    return processor_class