Skip to content

dandy

dandy

__all__ = ['Agent', 'BaseIntel', 'BaseListIntel', 'Bot', 'DandyCriticalException', 'DandyException', 'DandyRecoverableException', 'Decoder', 'LlmConfigOptions', 'MemoryCache', 'Prompt', 'Recorder', 'SqliteCache', 'cache_to_memory', 'cache_to_sqlite', 'generate_cache_key', 'process_to_future', 'recorder_to_html_file', 'recorder_to_json_file', 'recorder_to_markdown_file'] module-attribute

BaseIntel

Bases: BaseModel, ABC

model_to_kwargs

Source code in dandy/intel/intel.py 
24
25
def model_to_kwargs(self) -> dict:
    return dict(self)

model_inc_ex_class_copy classmethod

Source code in dandy/intel/intel.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def model_inc_ex_class_copy(
    cls,
    include: IncEx | dict | None = None,
    exclude: IncEx | dict | None = None,
    intel_object: Self | None = None,
) -> type[BaseIntel]:
    if include is None and exclude is None:
        return create_model(cls.__name__, __base__=cls)

    if include and exclude:
        message = "include and exclude cannot be used together"
        raise IntelCriticalException(message)

    include_dict = cls._inc_ex_to_dict(include)
    exclude_dict = cls._inc_ex_to_dict(exclude)

    cls._validate_inc_ex_dict_or_error(
        include_dict,
        exclude_dict,
    )

    cls._validate_inc_ex_value_or_error(
        include,
        include_dict,
        exclude,
        exclude_dict,
        intel_object,
    )

    processed_fields = {}

    for field_name, field_info in cls.model_fields.items():
        include_value = include_dict.get(field_name)
        exclude_value = exclude_dict.get(field_name)

        field_annotation = FieldAnnotation(field_info.annotation, field_name)

        field_default_value = cls._get_field_default_value_from_field_info(field_info)

        field_annotation_type = False

        if isinstance(include_value, dict) or isinstance(exclude_value, dict):
            field_annotation_origin = field_annotation.origin

            if field_annotation_origin is UnionType:
                field_annotation_origin = Union

            if issubclass(field_annotation.first_inner, BaseIntel):
                sub_model: type[BaseIntel] = field_annotation.first_inner

                new_sub_model = sub_model.model_inc_ex_class_copy(
                    include=include_value,
                    exclude=exclude_value,
                )

                field_annotation_type = (
                    new_sub_model
                    if field_annotation_origin is None
                    else field_annotation_origin[new_sub_model]
                )

            else:
                field_annotation_type = (
                    field_annotation.first_inner
                    if field_annotation_origin is None
                    else field_annotation_origin[field_annotation.first_inner]
                )

        elif (include_value and exclude is None) or (
            exclude_value is None and include is None
        ):
            field_annotation_type = field_annotation.base

        if field_annotation_type:
            processed_fields[field_name] = (
                field_annotation_type,
                field_default_value,
            )

    return create_model(cls.__name__, **processed_fields, __base__=BaseIntel)

model_json_inc_ex_schema classmethod

Source code in dandy/intel/intel.py 
109
110
111
112
113
114
115
116
117
118
@classmethod
def model_json_inc_ex_schema(
    cls,
    include: IncEx | None = None,
    exclude: IncEx | None = None,
) -> dict:
    return cls.model_inc_ex_class_copy(
        include=include,
        exclude=exclude,
    ).model_json_schema()

model_object_json_inc_ex_schema

Source code in dandy/intel/intel.py 
120
121
122
123
124
125
def model_object_json_inc_ex_schema(
    self, include: IncEx | None = None, exclude: IncEx | None = None
) -> dict:
    return self.model_inc_ex_class_copy(
        include=include, exclude=exclude, intel_object=self
    ).model_json_schema()

model_validate_and_copy

Source code in dandy/intel/intel.py 
127
128
129
130
def model_validate_and_copy(self, update: dict) -> Self:
    return self.model_validate(
        obj=self.model_copy(update=update).model_dump(warnings=False),
    )

model_validate_json_and_copy

Source code in dandy/intel/intel.py 
132
133
def model_validate_json_and_copy(self, json_data: str) -> Self:
    return self.model_validate_and_copy(update=from_json(json_data))

BaseListIntel

Bases: BaseIntel, ABC, Generic[T]

model_post_init

Source code in dandy/intel/intel.py 
214
215
216
217
218
219
220
221
222
223
224
225
def model_post_init(self, __context: Any, /):
    list_fields = [
        name
        for name, field in self.__class__.model_fields.items()
        if get_origin(field.annotation) is list
    ]

    if len(list_fields) != 1:
        message = "BaseListIntel sub classes can only have exactly one list field attribute and must be declared with typing"
        raise ValueError(message)

    self._list_name = list_fields[0]

__getitem__

Source code in dandy/intel/intel.py 
227
228
def __getitem__(self, index: int) -> list[T] | T:
    return getattr(self, self._list_name)[index]

__iter__

Source code in dandy/intel/intel.py 
230
231
def __iter__(self) -> Iterator[T]:
    yield from getattr(self, self._list_name)

__len__

Source code in dandy/intel/intel.py 
233
234
def __len__(self) -> int:
    return len(getattr(self, self._list_name))

__setitem__

Source code in dandy/intel/intel.py 
236
237
def __setitem__(self, index: int, value: T):
    getattr(self, self._list_name)[index] = value

append

Source code in dandy/intel/intel.py 
239
240
def append(self, item: T):
    getattr(self, self._list_name).append(item)

extend

Source code in dandy/intel/intel.py 
242
243
def extend(self, items: list[T]):
    getattr(self, self._list_name).extend(items)

Bot

Bases: BaseProcessor, BotServiceMixin, LlmServiceMixin, HttpServiceMixin, IntelServiceMixin, VisionServiceMixin

Source code in dandy/processor/bot/bot.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
        self,
        llm_randomize_seed: bool | None = None,
        llm_seed: int | None = None,
        llm_temperature: float | None = None,
        **kwargs
):
    super().__init__(
        **kwargs
    )

    self.llm_config_options.update_values(
        randomize_seed=llm_randomize_seed,
        seed=llm_seed,
        temperature=llm_temperature,
    )

description = 'Generic Bot for performing generic tasks' class-attribute instance-attribute

get_description classmethod

Source code in dandy/processor/bot/bot.py 
41
42
43
44
45
46
@classmethod
def get_description(cls) -> str | None:
    if cls.description is not None:
        return cls.description

    return cls.get_llm_description()

process

Source code in dandy/processor/bot/bot.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def process(
        self,
        *args,
        **kwargs,
) -> Any:
    if len(args) >= 1 and isinstance(args[0], PromptOrStr):
        kwargs['prompt'] = args[0]

    if len(args) == 2 and issubclass(args[1], BaseIntel):
        kwargs['intel_class'] = args[1]

    if 'prompt' in kwargs:
        return self.llm.prompt_to_intel(
            **kwargs
        )

    message = '`Bot.process` requires key word argument `prompt`.'
    raise ValueError(message)

Agent

Bases: BaseProcessor, AgentServiceMixin, LlmServiceMixin, HttpServiceMixin, IntelServiceMixin, VisionServiceMixin

Source code in dandy/processor/agent/agent.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
        self,
        llm_randomize_seed: bool | None = None,
        llm_seed: int | None = None,
        llm_temperature: float | None = None,
        processors: Sequence[type[BaseProcessor]] | None = None,
        **kwargs
):
    super().__init__(
        **generate_forwardable_kwargs_if_not_none(
            processors=processors
        ),
        **kwargs
    )

    self.llm_config_options.update_values(
        randomize_seed=llm_randomize_seed,
        seed=llm_seed,
        temperature=llm_temperature,
    )

plan_time_limit_seconds = settings.AGENT_DEFAULT_PLAN_TIME_LIMIT_SECONDS class-attribute instance-attribute

plan_task_count_limit = settings.AGENT_DEFAULT_PLAN_TASK_COUNT_LIMIT class-attribute instance-attribute

processors = (GenericTaskBot,) class-attribute instance-attribute

services = AgentService() class-attribute

__init_subclass__

Source code in dandy/processor/agent/agent.py 
72
73
74
75
76
77
78
79
80
def __init_subclass__(cls, **kwargs):
    if cls.processors is None or len(cls.processors) == 0:
        message = f'{cls.__name__} must have a sequence of "BaseProcessor" sub classes defined on the "processors" class attribute.'
        raise AgentCriticalException(message)

    for processor in cls.processors:
        if processor is Bot:
            message = f'{cls.__name__} cannot have a "Bot" class defined on the "processors" class attribute.'
            raise AgentCriticalException(message)

__post_init__

Source code in dandy/processor/agent/agent.py 
82
83
84
85
def __post_init__(self):
    self._processors_strategy = ProcessorsStrategy(
        self.processors
    )

get_description classmethod

Source code in dandy/processor/agent/agent.py 
87
88
89
90
91
92
@classmethod
def get_description(cls) -> str | None:
    if cls.description is not None:
        return cls.description

    return cls.get_llm_description()

process

Source code in dandy/processor/agent/agent.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def process(
        self,
        prompt: PromptOrStr,
        intel_class: type[IntelType] | None = None,
        intel_object: IntelType | None = None,
        images: list[str] | None = None,
        image_files: list[str | Path] | None = None,
        include_fields: IncEx | None = None,
        exclude_fields: IncEx | None = None,
        message_history: MessageHistory | None = None,
) -> IntelType:

    plan_intel = self._create_plan(prompt)

    completed_plan_intel = self._run_plan(plan_intel)

    recorder_add_llm_agent_processing_final_result_event(
        completed_plan_intel,
        self._recorder_event_id
    )

    self.llm.add_message(
        role='user',
        content=(
            Prompt()
            .text('I have the following task: ')
            .line_break()
            .text(prompt if isinstance(prompt, str) else prompt.to_str())
            .line_break()
            .text('Please create and execute a plan to accomplish this task.')
            .to_str()
        )
    )

    self.llm.add_message(
        role='assistant',
        content=(
            Prompt()
            .sub_heading('Plan Execution Results:')
            .line_break()
            .prompt(completed_plan_intel.to_prompt())
            .to_str()
        )
    )

    self.llm.add_message(
        role='user',
        content=(
            Prompt()
            .text('Use the results of the executed plan to complete: ')
            .line_break()
            .text(prompt if isinstance(prompt, str) else prompt.to_str())
            .to_str()
        )
    )

    return self.llm.prompt_to_intel(
        intel_class=intel_class,
        intel_object=intel_object,
        images=images,
        image_files=image_files,
        include_fields=include_fields,
        exclude_fields=exclude_fields,
        message_history=message_history,
    )

DandyException

Bases: Exception

DandyRecoverableException

Bases: DandyException

DandyCriticalException

Bases: DandyException

Decoder

Bases: BaseProcessor, DecoderServiceMixin, LlmServiceMixin

Source code in dandy/processor/decoder/decoder.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
        self,
        llm_randomize_seed: bool | None = None,
        llm_seed: int | None = None,
        llm_temperature: float | None = None,
        mapping: dict[str, Any] | None = None,
        mapping_keys_description: str | None = None,
        **kwargs
):
    super().__init__(
        **generate_forwardable_kwargs_if_not_none(
            mapping=mapping,
            mapping_keys_description=mapping_keys_description,
        ),
        **kwargs
    )

    self.llm_config_options.update_values(
        randomize_seed=llm_randomize_seed,
        seed=llm_seed,
        temperature=llm_temperature,
    )

mapping = None class-attribute instance-attribute

mapping_keys_description = None class-attribute instance-attribute

services = DecoderService() class-attribute

__init_subclass__

Source code in dandy/processor/decoder/decoder.py 
68
69
70
71
72
73
74
75
76
77
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 DecoderCriticalException(message)

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

__post_init__

Source code in dandy/processor/decoder/decoder.py 
79
80
81
82
83
84
85
86
87
88
89
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:
        if not isinstance(key, str):
            message = f'Decoder keys must be strings, found {key} ({type(key)}).'
            raise DecoderCriticalException(message)

__getitem__

Source code in dandy/processor/decoder/decoder.py 
91
92
def __getitem__(self, item: str) -> Any:
    return self.mapping[item]

as_enum

Source code in dandy/processor/decoder/decoder.py 
115
116
117
118
119
def as_enum(self) -> Enum:
    return Enum(
        f'{self.__class__.__name__}Enum',
        {value[0]: key for key, value in self._keyed_mapping.items()},
    )

get_description classmethod

Source code in dandy/processor/decoder/decoder.py 
127
128
129
130
131
132
@classmethod
def get_description(cls) -> str | None:
    if cls.description is not None:
        return cls.description

    return cls.get_llm_description()

process

Source code in dandy/processor/decoder/decoder.py 
134
135
136
137
138
139
def process(
    self,
    prompt: PromptOrStr,
    max_return_values: int | None = None,
) -> DecoderValuesIntel:
    return self._process_decoder_to_intel(prompt, max_return_values)

process_to_future

Source code in dandy/processor/decoder/decoder.py 
307
308
def process_to_future(self, *args, **kwargs) -> AsyncFuture[DecoderValuesIntel]:
    return AsyncFuture[DecoderValuesIntel](self.process, *args, **kwargs)

MemoryCache

Bases: BaseCache

cache_name instance-attribute

limit instance-attribute

__len__

Source code in dandy/cache/memory/cache.py 
20
21
def __len__(self) -> int:
    return len(self._cache)

get

Source code in dandy/cache/memory/cache.py 
23
24
def get(self, key: str) -> Any | None:
    return self._cache.get(key)

set

Source code in dandy/cache/memory/cache.py 
26
27
28
def set(self, key: str, value: Any):
    self._cache[key] = value
    self.clean()

clean

Source code in dandy/cache/memory/cache.py 
30
31
32
def clean(self):
    if len(self._cache) > self.limit:
        self._cache.popitem(last=False)

clear classmethod

Source code in dandy/cache/memory/cache.py 
34
35
36
37
@classmethod
def clear(cls, cache_name: str = dandy.consts.CACHE_DEFAULT_NAME):
    if cache_name in _memory_cache:
        _memory_cache[cache_name].clear()

clear_all classmethod

Source code in dandy/cache/memory/cache.py 
39
40
41
@classmethod
def clear_all(cls):
    _memory_cache.clear()

destroy_all classmethod

Source code in dandy/cache/memory/cache.py 
43
44
45
@classmethod
def destroy_all(cls):
    cls.clear_all()

Prompt dataclass

input = (None,) class-attribute instance-attribute

tag = (None,) class-attribute instance-attribute

estimated_token_count property

__post_init__

Source code in dandy/llm/prompt/prompt.py 
17
18
19
20
21
22
23
24
25
26
def __post_init__(
    self,
):
    self.snippets: list[snippet.BaseSnippet] = []

    if isinstance(self.input, Prompt):
        self.text(text=self.input.to_str())

    if isinstance(self.input, str):
        self.text(text=self.input)

__str__

Source code in dandy/llm/prompt/prompt.py 
28
29
def __str__(self) -> str:
    return self.to_str()

to_str

Source code in dandy/llm/prompt/prompt.py 
31
32
33
34
35
36
37
def to_str(self) -> str:
    prompt_string = ''.join([snippet_.to_str() for snippet_ in self.snippets])

    if isinstance(self.tag, str):
        return f'<{self.tag}>\n{prompt_string}\n</{self.tag}>\n'

    return prompt_string

dict

Source code in dandy/llm/prompt/prompt.py 
39
40
41
42
43
44
def dict(self, dictionary: dict, triple_quote: bool = False) -> Self:
    self.snippets.append(
        snippet.DictionarySnippet(dictionary=dictionary, triple_quote=triple_quote)
    )

    return self

directory_list

Source code in dandy/llm/prompt/prompt.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def directory_list(
    self,
    directory_path: str | Path,
    max_depth: int | None = None,
    file_extensions: Sequence[str] | None = None,
    triple_quote: bool = False,
) -> Self:
    self.snippets.append(
        snippet.DirectoryListSnippet(
            directory_path=directory_path,
            triple_quote=triple_quote,
            max_depth=max_depth,
            file_extensions=file_extensions,
        )
    )

    return self

divider

Source code in dandy/llm/prompt/prompt.py 
64
65
66
67
def divider(self) -> Self:
    self.snippets.append(snippet.DividerSnippet())

    return self

array

Source code in dandy/llm/prompt/prompt.py 
69
70
71
72
def array(self, items: list[str]) -> Self:
    self.snippets.append(snippet.ArraySnippet(items=items))

    return self

array_random_order

Source code in dandy/llm/prompt/prompt.py 
74
75
76
77
def array_random_order(self, items: list[str]) -> Self:
    self.snippets.append(snippet.ArrayRandomOrderSnippet(items=items))

    return self

file

Source code in dandy/llm/prompt/prompt.py 
83
84
85
86
87
88
89
90
91
92
def file(
    self, file_path: str | Path, encoding: str = 'utf-8', triple_quote: bool = False
) -> Self:
    self.snippets.append(
        snippet.FileSnippet(
            file_path=file_path, encoding=encoding, triple_quote=triple_quote
        )
    )

    return self

heading

Source code in dandy/llm/prompt/prompt.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def heading(
    self,
    heading: str,
) -> Self:
    self.snippets.append(
        snippet.HeadingSnippet(
            heading=heading,
        )
    )

    return self

line_break

Source code in dandy/llm/prompt/prompt.py 
106
107
108
109
def line_break(self) -> Self:
    self.snippets.append(snippet.LineBreakSnippet())

    return self

list

Source code in dandy/llm/prompt/prompt.py 
111
112
113
114
def list(self, items: list[str], triple_quote: bool = False) -> Self:
    self.unordered_list(items=items, triple_quote=triple_quote)

    return self

intel

Source code in dandy/llm/prompt/prompt.py 
116
117
118
119
120
121
def intel(self, intel: BaseIntel, triple_quote: bool = False) -> Self:
    self.snippets.append(
        snippet.IntelSnippet(intel=intel, triple_quote=triple_quote)
    )

    return self

intel_schema

Source code in dandy/llm/prompt/prompt.py 
123
124
125
126
127
128
129
130
131
132
def intel_schema(
    self, intel_class: type[BaseIntel], triple_quote: bool = False
) -> Self:
    self.snippets.append(
        snippet.IntelSchemaSnippet(
            intel_class=intel_class, triple_quote=triple_quote
        )
    )

    return self

module_source

Source code in dandy/llm/prompt/prompt.py 
134
135
136
137
138
139
140
141
142
143
def module_source(self, module_name: str, triple_quote: bool = True) -> Self:
    self.snippets.append(
        snippet.ModuleSourceSnippet(
            module_name=module_name,
            triple_quote=triple_quote,
            triple_quote_label=module_name,
        )
    )

    return self

object_source

Source code in dandy/llm/prompt/prompt.py 
145
146
147
148
149
150
151
152
153
154
def object_source(self, object_module_name: str, triple_quote: bool = True) -> Self:
    self.snippets.append(
        snippet.ObjectSourceSnippet(
            object_module_name=object_module_name,
            triple_quote=triple_quote,
            triple_quote_label=object_module_name,
        )
    )

    return self

ordered_list

Source code in dandy/llm/prompt/prompt.py 
156
157
158
159
160
161
162
163
def ordered_list(
    self, items: builtins.list[str], triple_quote: bool = False
) -> Self:
    self.snippets.append(
        snippet.OrderedListSnippet(items=items, triple_quote=triple_quote)
    )

    return self

prompt

Source code in dandy/llm/prompt/prompt.py 
165
166
167
168
169
170
def prompt(self, prompt: Self | str, triple_quote: bool = False) -> Self:
    self.snippets.append(
        snippet.PromptSnippet(prompt=prompt, triple_quote=triple_quote)
    )

    return self

random_choice

Source code in dandy/llm/prompt/prompt.py 
172
173
174
175
176
177
178
179
180
181
182
def random_choice(
    self, choices: builtins.list[str], triple_quote: bool = False
) -> Self:
    self.snippets.append(
        snippet.RandomChoiceSnippet(
            choices=choices,
            triple_quote=triple_quote,
        )
    )

    return self

sub_heading

Source code in dandy/llm/prompt/prompt.py 
184
185
186
187
188
189
190
191
192
193
194
def sub_heading(
    self,
    sub_heading: str,
) -> Self:
    self.snippets.append(
        snippet.SubHeadingSnippet(
            sub_heading=sub_heading,
        )
    )

    return self

text

Source code in dandy/llm/prompt/prompt.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def text(
    self,
    text: str = '',
    label: str = '',
    triple_quote: bool = False,
    triple_quote_label: str | None = None,
) -> Self:
    self.snippets.append(
        snippet.TextSnippet(
            text=text,
            label=label,
            triple_quote=triple_quote,
            triple_quote_label=triple_quote_label,
        )
    )

    return self

title

Source code in dandy/llm/prompt/prompt.py 
214
215
216
217
218
219
220
221
222
223
224
def title(
    self,
    title: str,
) -> Self:
    self.snippets.append(
        snippet.TitleSnippet(
            title=title,
        )
    )

    return self

unordered_list

Source code in dandy/llm/prompt/prompt.py 
226
227
228
229
230
231
232
233
def unordered_list(
    self, items: builtins.list[str], triple_quote: bool = False
) -> Self:
    self.snippets.append(
        snippet.UnorderedListSnippet(items=items, triple_quote=triple_quote)
    )

    return self

unordered_random_list

Source code in dandy/llm/prompt/prompt.py 
235
236
237
238
239
240
241
242
243
244
245
246
def unordered_random_list(
    self, items: builtins.list[str], triple_quote: bool = False
) -> Self:

    self.snippets.append(
        snippet.UnorderedRandomListSnippet(
            items=items,
            triple_quote=triple_quote
        )
    )

    return self

LlmConfigOptions

Source code in dandy/llm/config/options.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def __init__(
    self,
    seed: int | None = None,
    randomize_seed: bool | None = None,
    max_input_tokens: int | None = None,
    max_output_tokens: int | None = None,
    temperature: float | None = None,
    prompt_retry_count: int | None = None,
):
    self._seed = seed
    self._randomize_seed = randomize_seed
    self._max_input_tokens = max_input_tokens
    self._max_output_tokens = max_output_tokens
    self._temperature = temperature
    self._prompt_retry_count = prompt_retry_count

seed property

randomize_seed property

max_input_tokens property

max_output_tokens property

temperature property

prompt_retry_count property

update_values

Source code in dandy/llm/config/options.py 
71
72
73
74
def update_values(self, **kwargs):
    for key, value in kwargs.items():
        if hasattr(self, key) and value is not None:
            setattr(self, key, value)

merge_to_copy

Merges the current instance with another secondary instance Current instance attributes that are not none will take precedence over the secondary instance

Source code in dandy/llm/config/options.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def merge_to_copy(self, secondary_options: Self) -> Self:
    """
    Merges the current instance with another secondary instance
    Current instance attributes that are not none will take precedence over the secondary instance
    """

    merged_dict = {
        **{
            key: value
            for key, value in secondary_options.__dict__.items()
            if value is not None
        },
        **{key: value for key, value in self.__dict__.items() if value is not None},
    }

    return self.__class__(
        **{
            key.removeprefix('_'): value
            for key, value in merged_dict.items()
        }
    )

Recorder

Bases: Singleton

recordings = {} class-attribute instance-attribute

renderers = {'html': HtmlRecordingRenderer, 'json': JsonRecordingRenderer, 'markdown': MarkdownRecordingRenderer} class-attribute instance-attribute

add_event classmethod

Source code in dandy/recorder/recorder.py 
27
28
29
30
31
@classmethod
def add_event(cls, event: Event):
    for recording in cls.recordings.values():
        if recording.is_running:
            recording.event_store.add_event(event)

check_recording_is_valid classmethod

Source code in dandy/recorder/recorder.py 
33
34
35
36
37
38
39
40
41
42
@classmethod
def check_recording_is_valid(cls, recording_name: str = RECORDING_DEFAULT_NAME):
    if recording_name not in cls.recordings:
        choices_message = ''

        if len(cls.recordings.keys()) == 0:
            choices_message = f' Choices are {list(cls.recordings.keys())}'

        message = f'Recording "{recording_name}" does not exist. {choices_message}'
        raise RecorderCriticalException(message)

delete_all_recordings classmethod

Source code in dandy/recorder/recorder.py 
44
45
46
@classmethod
def delete_all_recordings(cls):
    cls.recordings.clear()

delete_recording classmethod

Source code in dandy/recorder/recorder.py 
48
49
50
51
@classmethod
def delete_recording(cls, recording_name: str = RECORDING_DEFAULT_NAME):
    cls.check_recording_is_valid(recording_name)
    del cls.recordings[recording_name]

get_recording classmethod

Source code in dandy/recorder/recorder.py 
53
54
55
56
@classmethod
def get_recording(cls, recording_name: str = RECORDING_DEFAULT_NAME) -> Recording:
    cls.check_recording_is_valid(recording_name)
    return cls.recordings[recording_name]

is_recording classmethod

Source code in dandy/recorder/recorder.py 
58
59
60
61
62
@classmethod
def is_recording(cls):
    return any(
        recording.is_running for recording in cls.recordings.values()
    )

start_recording classmethod

Source code in dandy/recorder/recorder.py 
64
65
66
67
@classmethod
def start_recording(cls, recording_name: str = RECORDING_DEFAULT_NAME):
    cls.recordings[recording_name] = Recording(name=recording_name)
    cls.recordings[recording_name].start()

stop_recording classmethod

Source code in dandy/recorder/recorder.py 
69
70
71
72
@classmethod
def stop_recording(cls, recording_name: str = RECORDING_DEFAULT_NAME):
    cls.check_recording_is_valid(recording_name)
    cls.recordings[recording_name].stop()

stop_all_recording classmethod

Source code in dandy/recorder/recorder.py 
74
75
76
77
@classmethod
def stop_all_recording(cls):
    for recording in cls.recordings.values():
        recording.stop()

to_file classmethod

Source code in dandy/recorder/recorder.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def to_file(
        cls,
        recording_name: str,
        renderer: str,
        path: Path | str
):
    return cls._render(
        to_file=True,
        renderer=renderer,
        recording_name=recording_name,
        path=path
    )

to_html_file classmethod

Source code in dandy/recorder/recorder.py 
130
131
132
133
134
135
136
137
138
139
140
@classmethod
def to_html_file(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
        path: Path | str = DEFAULT_RECORDER_OUTPUT_PATH
):
    cls.to_file(
        recording_name,
        'html',
        path
    )

to_html_str classmethod

Source code in dandy/recorder/recorder.py 
142
143
144
145
146
147
148
149
150
@classmethod
def to_html_str(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
) -> str:
    return cls._to_str(
        recording_name,
        'html',
    )

to_json_file classmethod

Source code in dandy/recorder/recorder.py 
152
153
154
155
156
157
158
159
160
161
162
@classmethod
def to_json_file(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
        path: Path | str = DEFAULT_RECORDER_OUTPUT_PATH
):
    cls.to_file(
        recording_name,
        'json',
        path
    )

to_json_str classmethod

Source code in dandy/recorder/recorder.py 
164
165
166
167
168
169
170
171
172
@classmethod
def to_json_str(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
) -> str:
    return cls._to_str(
        recording_name,
        'json',
    )

to_markdown_file classmethod

Source code in dandy/recorder/recorder.py 
174
175
176
177
178
179
180
181
182
183
184
@classmethod
def to_markdown_file(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
        path: Path | str = DEFAULT_RECORDER_OUTPUT_PATH
):
    cls.to_file(
        recording_name,
        'markdown',
        path
    )

to_markdown_str classmethod

Source code in dandy/recorder/recorder.py 
186
187
188
189
190
191
192
193
194
@classmethod
def to_markdown_str(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
) -> str:
    return cls._to_str(
        recording_name,
        'markdown',
    )

SqliteCache

Bases: BaseCache

cache_name instance-attribute

limit instance-attribute

model_post_init

Source code in dandy/cache/sqlite/cache.py 
15
16
def model_post_init(self, __context: Any, /):
    self._create_table()

__len__

Source code in dandy/cache/sqlite/cache.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
def __len__(self) -> int:
    if not self._table_exists():
        return 0

    with SqliteConnection(SQLITE_CACHE_DB_NAME) as connection:
        cursor = connection.cursor()

        cursor.execute(
            f'SELECT COUNT(*) FROM {SQLITE_CACHE_TABLE_NAME} WHERE cache_name = ?',
            (self.cache_name,)
        )

        return cursor.fetchone()[0]

get

Source code in dandy/cache/sqlite/cache.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def get(self, key: str) -> Any | None:
    if not self._table_exists():
        return None

    with SqliteConnection(SQLITE_CACHE_DB_NAME) as connection:
        cursor = connection.cursor()

        cursor.execute(
            f'SELECT value FROM {SQLITE_CACHE_TABLE_NAME} WHERE key = ? AND cache_name = ?',
            (key, self.cache_name)
        )
        result = cursor.fetchone()

        if result:
            return pickle.loads(result[0])

        return None

set

Source code in dandy/cache/sqlite/cache.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def set(self, key: str, value: Any):
    with SqliteConnection(SQLITE_CACHE_DB_NAME) as connection:
        cursor = connection.cursor()

        value_string = pickle.dumps(value)

        try:
            cursor.execute(
                f'INSERT INTO {SQLITE_CACHE_TABLE_NAME} (key, value, cache_name) VALUES (?, ?, ?)',
                (key, value_string, self.cache_name)
            )
        except sqlite3.IntegrityError:
            cursor.execute(
                f'UPDATE {SQLITE_CACHE_TABLE_NAME} SET value = ? WHERE key = ? AND cache_name = ?',
                (value_string, key, self.cache_name)
            )

        connection.commit()
        self.clean()

clean

Source code in dandy/cache/sqlite/cache.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def clean(self):
    with SqliteConnection(SQLITE_CACHE_DB_NAME) as connection:
        cursor = connection.cursor()

        excess_threshold = int(self.limit * 0.10)
        excess_rows = self.__len__() - self.limit

        if excess_rows >= excess_threshold:
            cursor.execute(
                f'DELETE FROM {SQLITE_CACHE_TABLE_NAME} WHERE key IN (SELECT key FROM {SQLITE_CACHE_TABLE_NAME} WHERE cache_name = ? ORDER BY created_at LIMIT ?)',
                (self.cache_name, excess_threshold)
            )

        connection.commit()

clear classmethod

Source code in dandy/cache/sqlite/cache.py 
110
111
112
113
114
115
116
117
118
119
@classmethod
def clear(cls, cache_name: str = dandy.consts.CACHE_DEFAULT_NAME):
    if cls._table_exists():
        with SqliteConnection(SQLITE_CACHE_DB_NAME) as connection:
            cursor = connection.cursor()
            cursor.execute(
                f'DELETE FROM {SQLITE_CACHE_TABLE_NAME} WHERE cache_name = ?',
                (cache_name,)
            )
            connection.commit()

clear_all classmethod

Source code in dandy/cache/sqlite/cache.py 
121
122
123
124
125
126
127
@classmethod
def clear_all(cls):
    if cls._table_exists():
        with SqliteConnection(SQLITE_CACHE_DB_NAME) as connection:
            cursor = connection.cursor()
            cursor.execute(f'DELETE FROM {SQLITE_CACHE_TABLE_NAME}')
            connection.commit()

destroy_all classmethod

Source code in dandy/cache/sqlite/cache.py 
129
130
131
@classmethod
def destroy_all(cls):
    SqliteConnection(SQLITE_CACHE_DB_NAME).delete_db_file()

generate_cache_key

Source code in dandy/cache/tools.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def generate_cache_key(func: object, *args, **kwargs) -> str:
    hashable_args = tuple(
        convert_to_hashable_str(arg) for arg in args
    )

    hashable_kwargs = tuple(
        (key, convert_to_hashable_str(value)) for key, value in kwargs.items()
    )

    hashable_tuple = (
        func.__module__,
        func.__qualname__,
        hashable_args,
        hashable_kwargs,
    )

    hash_key = hashlib.shake_128(
        str(hashable_tuple).encode()
    ).hexdigest(16)

    return hash_key

process_to_future

Source code in dandy/core/future/tools.py 
 8
 9
10
11
12
13
def process_to_future(
        callable_: Callable[..., R],
        *args,
        **kwargs
) -> AsyncFuture[R]:
    return AsyncFuture[R](callable_, *args, **kwargs)

cache_to_memory

Source code in dandy/cache/memory/decorators.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def cache_to_memory(
        cache_name: str = dandy.consts.CACHE_DEFAULT_NAME,
        limit: int = settings.CACHE_MEMORY_LIMIT
) -> Callable:
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Callable:
            return cache_decorator_function(
                MemoryCache(
                    cache_name=cache_name,
                    limit=limit
                ),
                func,
                *args,
                **kwargs
            )

        return wrapper

    return decorator

cache_to_sqlite

Source code in dandy/cache/sqlite/decorators.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def cache_to_sqlite(
        cache_name: str = dandy.consts.CACHE_DEFAULT_NAME,
        limit: int = settings.CACHE_SQLITE_LIMIT
) -> Callable:
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Callable:
            return cache_decorator_function(
                SqliteCache(
                    cache_name=cache_name,
                    limit=limit,
                ),
                func,
                *args,
                **kwargs
            )

        return wrapper

    return decorator

recorder_to_html_file

Source code in dandy/recorder/decorators.py 
38
39
40
41
42
43
44
45
46
def recorder_to_html_file(recording_name: str | None = None, path: Path | str = DEFAULT_RECORDER_OUTPUT_PATH):
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return _recorder_to_file_decorator_function(func, args, kwargs, recording_name, 'html', path)

        return wrapper

    return decorator

recorder_to_json_file

Source code in dandy/recorder/decorators.py 
49
50
51
52
53
54
55
56
57
def recorder_to_json_file(recording_name: str | None = None, path: Path | str = DEFAULT_RECORDER_OUTPUT_PATH):
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return _recorder_to_file_decorator_function(func, args, kwargs, recording_name, 'json', path)

        return wrapper

    return decorator

recorder_to_markdown_file

Source code in dandy/recorder/decorators.py 
60
61
62
63
64
65
66
67
68
def recorder_to_markdown_file(recording_name: str | None = None, path: Path | str = DEFAULT_RECORDER_OUTPUT_PATH):
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return _recorder_to_file_decorator_function(func, args, kwargs, recording_name, 'markdown', path)

        return wrapper

    return decorator