class documentation

class TopologicalProcessor(abc.ABC, Generic[KT, VT, RT]): (source)

View In Hierarchy

Base class for processing objects in topological order. Decoupled from the concrete types so it can be re-used for several order-sensitive analysis.

Base classes must override getObj and processObj.

The processing code can call getProcessedObj to indicate that the given object should be processed before resuming the current task. In case of cycles, getProcessedObj returns a object that is not completely processed. To know wether the returned obeject is processed, one can use the processing_state mapping or the result mapping, a object that is still processing will not be present in the result mapping.

Method __init__ Undocumented
Method getObj Returns the object known by the given key.
Method getProcessedObj Request processing of the given object and returns the object. The object might still be processing in the case of cycles.
Method process Run the processObj method on all objects.
Method processObj Process the object known by the given key and returns the 'result' of the analysis. If the processing code mutates a global state, there might be no need for the return value.
Constant PROCESSED Undocumented
Constant PROCESSING Undocumented
Constant UNPROCESSED Undocumented
Instance Variable processing Undocumented
Instance Variable processing_state Undocumented
Instance Variable result Undocumented
Instance Variable unprocessed Undocumented
Method _processObj Undocumented
def __init__(self): (source)

Undocumented

@abc.abstractmethod
def getObj(self, key: KT) -> VT | None: (source)

Returns the object known by the given key.

def getProcessedObj(self, key: KT) -> VT | None: (source)

Request processing of the given object and returns the object. The object might still be processing in the case of cycles.

def process(self, objects: Iterable[VT]) -> dict[VT, RT]: (source)

Run the processObj method on all objects.

@abc.abstractmethod
def processObj(self, obj: VT) -> RT: (source)

Process the object known by the given key and returns the 'result' of the analysis. If the processing code mutates a global state, there might be no need for the return value.

PROCESSED: int = (source)

Undocumented

Value
2
PROCESSING: int = (source)

Undocumented

Value
1
UNPROCESSED: int = (source)

Undocumented

Value
0
processing: list[VT] = (source)

Undocumented

processing_state: dict[VT, object] = (source)

Undocumented

result: dict[VT, RT] = (source)

Undocumented

unprocessed: list[VT] = (source)

Undocumented

def _processObj(self, obj: VT): (source)

Undocumented