Processor

BaseProcessor

Bases: ABC

Base class for all processing classes in dandy.

Source code in dandy/processor/processor.py

class BaseProcessor(ABC, metaclass=ProcessorABCMeta):
    """
    Base class for all processing classes in dandy.
    """
    @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
        """
        ...

    @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)

process(*args, **kwargs) abstractmethod classmethod

This method has hooks on it to allow for easy debugging

Parameters:

Name	Type	Description	Default
args		Arguments	()
kwargs		Keyword Arguments	{}

Returns:

Type	Description
Any	Any

Source code in dandy/processor/processor.py

@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
    """
    ...

process_to_future(*args, **kwargs) classmethod

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

Parameters:

Name	Type	Description	Default
args		Arguments	()
kwargs		Keyword Arguments	{}

Returns:

Type	Description
AsyncFuture	AsyncFuture

Source code in dandy/processor/processor.py

@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)

ProcessorABCMeta

Bases: ABCMeta

Source code in dandy/processor/abc_meta.py

class ProcessorABCMeta(ABCMeta):
    def __new__(cls, name, bases, dct):
        if 'process' in dct:
            original_process = dct['process']

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

                def wrapped_process(cls, *args, **kwargs):
                    if getattr(cls, "_debugger_called", None) is None:
                        cls._debugger_event_id = generate_new_debug_event_id()

                    if DebugRecorder.is_recording and not getattr(cls, "_debugger_called", False):
                        DebugRecorder.add_event(
                            RunEvent(
                                actor=pascal_to_title_case(cls.__name__),
                                action='Process',
                                description=json.dumps(
                                    {
                                        'args': args,
                                        'kwargs': kwargs
                                    },
                                    indent=4,
                                    default=json_default
                                ),
                                id=cls._debugger_event_id
                            )
                        )
                        cls._debugger_called = True

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

                    if DebugRecorder.is_recording and getattr(cls, "_debugger_called", True):
                        DebugRecorder.add_event(
                            ResultEvent(
                                actor=pascal_to_title_case(cls.__name__),
                                action='Process Returned Result',
                                description=json.dumps(
                                    {
                                        'returned result': result
                                    },
                                    indent=4,
                                    default=json_default
                                ),
                                id=cls._debugger_event_id
                            )
                        )

                    cls._debugger_called = False

                    return result

                dct['process'] = classmethod(wrapped_process)

        return super().__new__(cls, name, bases, dct)