Skip to content

dandy

dandy

__all__ = ['BaseIntel', 'BaseListIntel', 'Bot', 'DandyCriticalError', 'DandyError', 'DandyRecoverableError', '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

create_from_file classmethod

Source code in dandy/intel/intel.py 
29
30
31
@classmethod
def create_from_file(cls, file_path: Path | str) -> Self:
    return cls.model_validate_json(json_data=read_from_file(file_path=file_path))

model_to_kwargs

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

model_inc_ex_class_copy classmethod

Source code in dandy/intel/intel.py 
 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
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
@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 IntelCriticalError(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 
139
140
141
142
143
144
145
146
147
148
@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 
150
151
152
153
154
155
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 
157
158
159
160
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 
162
163
def model_validate_json_and_copy(self, json_data: str) -> Self:
    return self.model_validate_and_copy(update=from_json(json_data))

save_to_file

Source code in dandy/intel/intel.py 
165
166
def save_to_file(self, file_path: Path | str):
    write_to_file(file_path=file_path, content=self.model_dump_json(indent=2))

BaseListIntel

Bases: BaseIntel, ABC, Generic[T]

model_post_init

Source code in dandy/intel/intel.py 
235
236
237
238
239
240
241
242
243
244
245
246
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 
248
249
def __getitem__(self, index: int) -> list[T] | T:
    return getattr(self, self._list_name)[index]

__iter__

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

__len__

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

__setitem__

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

append

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

extend

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

Bot

Bases: FileServiceMixin, LlmServiceMixin, HttpServiceMixin, IntelServiceMixin

Source code in dandy/bot/bot.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    llm_config: str | None = None,
    llm_temperature: float | None = None,
    **kwargs,
):
    super().__init__(
        llm_config=llm_config,
        llm_temperature=llm_temperature,
        **kwargs,
    )

    self.recorder_event_id = ''
    self._recorder_called = None

    for key, value in kwargs.items():
        setattr(self, key, value)

    self.__post_init__()

recorder_event_id = '' instance-attribute

__init_subclass__

Source code in dandy/bot/bot.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init_subclass__(cls):
    super().__init_subclass__()

    if ABC not in cls.__bases__:
        # Typing Does not work properly for processors if you override __getattribute__ in the BaseProcessor class.
        # This is a workaround and should be fixed in future versions of the python lsp.
        def __getattribute__(self: Self, name: str) -> Any:  # noqa: N807
            attr = super().__getattribute__(name)

            if (
                name == 'process'
                and callable(attr)
                and not hasattr(attr, '_wrapped')
            ):
                wrapped = record_process_wrapper(self, attr)
                wrapped._wrapped = True

                return wrapped

            return attr

        cls.__getattribute__ = __getattribute__

__post_init__

Source code in dandy/bot/bot.py 
63
64
def __post_init__(self):  # noqa: B027
    pass

get_description classmethod

Source code in dandy/bot/bot.py 
66
67
68
@classmethod
def get_description(cls) -> str | None:
    pass

process

Source code in dandy/bot/bot.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
def process(
    self,
    *args,
    **kwargs,
) -> Any:
    if len(args) >= 1 and isinstance(args[0], Prompt | str):
        kwargs['prompt'] = args[0]

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

    message = '`Bot.process` requires `prompt` as an argument.'
    raise ValueError(message)

process_to_future

Source code in dandy/bot/bot.py 
84
85
def process_to_future(self, *args, **kwargs) -> AsyncFuture:
    return process_to_future(self.process, *args, **kwargs)

reset

Source code in dandy/bot/bot.py 
87
88
def reset(self):
    super().reset()

DandyError

Bases: Exception

DandyRecoverableError

Bases: DandyError

DandyCriticalError

Bases: DandyError

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.constants.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_backtick: bool = False) -> Self:
    self.snippets.append(
        snippet.DictionarySnippet(dictionary=dictionary, triple_backtick=triple_backtick)
    )

    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_backtick: bool = False,
) -> Self:
    self.snippets.append(
        snippet.DirectoryListSnippet(
            directory_path=directory_path,
            triple_backtick=triple_backtick,
            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
93
94
95
96
97
98
def file(
        self, file_path: str | Path,
        encoding: str = 'utf-8',
        triple_backtick: bool = False,
        triple_backtick_label: str | None = None
) -> Self:
    self.snippets.append(
        snippet.FileSnippet(
            file_path=file_path,
            encoding=encoding,
            triple_backtick=triple_backtick,
            triple_backtick_label=triple_backtick_label
        )
    )

    return self

heading

Source code in dandy/llm/prompt/prompt.py 
100
101
102
103
104
105
106
107
108
109
110
def heading(
        self,
        heading: str,
) -> Self:
    self.snippets.append(
        snippet.HeadingSnippet(
            heading=heading,
        )
    )

    return self

lb

Source code in dandy/llm/prompt/prompt.py 
112
113
def lb(self) -> Self:
    return self.line_break()

line_break

Source code in dandy/llm/prompt/prompt.py 
115
116
117
118
def line_break(self) -> Self:
    self.snippets.append(snippet.LineBreakSnippet())

    return self

list

Source code in dandy/llm/prompt/prompt.py 
120
121
122
123
def list(self, items: list[str], triple_backtick: bool = False) -> Self:
    self.unordered_list(items=items, triple_backtick=triple_backtick)

    return self

intel

Source code in dandy/llm/prompt/prompt.py 
125
126
127
128
129
130
def intel(self, intel: BaseIntel, triple_backtick: bool = False) -> Self:
    self.snippets.append(
        snippet.IntelSnippet(intel=intel, triple_backtick=triple_backtick)
    )

    return self

intel_schema

Source code in dandy/llm/prompt/prompt.py 
132
133
134
135
136
137
138
139
140
141
def intel_schema(
        self, intel_class: type[BaseIntel], triple_backtick: bool = False
) -> Self:
    self.snippets.append(
        snippet.IntelSchemaSnippet(
            intel_class=intel_class, triple_backtick=triple_backtick
        )
    )

    return self

module_source

Source code in dandy/llm/prompt/prompt.py 
143
144
145
146
147
148
149
150
151
152
def module_source(self, module_name: str, triple_backtick: bool = True) -> Self:
    self.snippets.append(
        snippet.ModuleSourceSnippet(
            module_name=module_name,
            triple_backtick=triple_backtick,
            triple_backtick_label=module_name,
        )
    )

    return self

object_source

Source code in dandy/llm/prompt/prompt.py 
154
155
156
157
158
159
160
161
162
163
def object_source(self, object_module_name: str, triple_backtick: bool = True) -> Self:
    self.snippets.append(
        snippet.ObjectSourceSnippet(
            object_module_name=object_module_name,
            triple_backtick=triple_backtick,
            triple_backtick_label=object_module_name,
        )
    )

    return self

ordered_list

Source code in dandy/llm/prompt/prompt.py 
165
166
167
168
169
170
171
172
def ordered_list(
        self, items: builtins.list[str], triple_backtick: bool = False
) -> Self:
    self.snippets.append(
        snippet.OrderedListSnippet(items=items, triple_backtick=triple_backtick)
    )

    return self

prompt

Source code in dandy/llm/prompt/prompt.py 
174
175
176
177
178
179
def prompt(self, prompt: Self | str, triple_backtick: bool = False) -> Self:
    self.snippets.append(
        snippet.PromptSnippet(prompt=prompt, triple_backtick=triple_backtick)
    )

    return self

random_choice

Source code in dandy/llm/prompt/prompt.py 
181
182
183
184
185
186
187
188
189
190
191
def random_choice(
        self, choices: builtins.list[str], triple_backtick: bool = False
) -> Self:
    self.snippets.append(
        snippet.RandomChoiceSnippet(
            choices=choices,
            triple_backtick=triple_backtick,
        )
    )

    return self

sub_heading

Source code in dandy/llm/prompt/prompt.py 
193
194
195
196
197
198
199
200
201
202
203
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 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def text(
        self,
        text: str = '',
        label: str = '',
        triple_backtick: bool = False,
        triple_backtick_label: str | None = None,
) -> Self:
    self.snippets.append(
        snippet.TextSnippet(
            text=text,
            label=label,
            triple_backtick=triple_backtick,
            triple_backtick_label=triple_backtick_label,
        )
    )

    return self

title

Source code in dandy/llm/prompt/prompt.py 
223
224
225
226
227
228
229
230
231
232
233
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 
235
236
237
238
239
240
241
242
def unordered_list(
        self, items: builtins.list[str], triple_backtick: bool = False
) -> Self:
    self.snippets.append(
        snippet.UnorderedListSnippet(items=items, triple_backtick=triple_backtick)
    )

    return self

unordered_random_list

Source code in dandy/llm/prompt/prompt.py 
244
245
246
247
248
249
250
251
252
253
254
255
def unordered_random_list(
        self, items: builtins.list[str], triple_backtick: bool = False
) -> Self:

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

    return self

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 RecorderCriticalError(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.constants.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 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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.constants.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.constants.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