Skip to content

map

dandy.processor.map.map

Map dataclass

Bases: BaseProcessor, LlmServiceMixin

mapping_keys_description = None class-attribute instance-attribute

mapping = None class-attribute instance-attribute

services = MapService() class-attribute

__init_subclass__

Source code in dandy/processor/map/map.py 
54
55
56
57
58
59
60
61
62
63
def __init_subclass__(cls):
    super().__init_subclass__()

    if cls.mapping_keys_description is None:
        message = f'{cls.__name__} `mapping_keys_description` is not set.'
        raise MapCriticalException(message)

    if cls.mapping is None:
        message = f'{cls.__name__} `mapping` is not set.'
        raise MapCriticalException(message)

__post_init__

Source code in dandy/processor/map/map.py 
65
66
67
68
69
70
71
72
73
74
75
def __post_init__(self):
    if self.mapping_keys_description is None:
        self.mapping_keys_description = self.__class__.mapping_keys_description

    if self.mapping is None:
        self.mapping = self.__class__.mapping

    for key in self.mapping.keys():
        if not isinstance(key, str):
            message = f'Mapping keys must be strings, found {key} ({type(key)}).'
            raise MapCriticalException(message)

__getitem__

Source code in dandy/processor/map/map.py 
77
78
def __getitem__(self, item):
    return self.mapping[item]

as_enum

Source code in dandy/processor/map/map.py 
80
81
82
83
84
85
86
87
def as_enum(self) -> Enum:
    return Enum(
        f'{self.__class__.__name__}Enum',
        {
            value[0]: key
            for key, value in self._keyed_mapping.items()
        }
    )

process

Source code in dandy/processor/map/map.py 
 95
 96
 97
 98
 99
100
101
102
103
def process(
        self,
        prompt: PromptOrStr,
        max_return_values: int | None = None,
) -> MapValuesIntel:
    return self._process_map_to_intel(
        prompt,
        max_return_values
    )

process_to_future

Source code in dandy/processor/map/map.py 
258
259
def process_to_future(self, *args, **kwargs) -> AsyncFuture[MapValuesIntel]:
    return AsyncFuture[MapValuesIntel](self.process, *args, **kwargs)