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

Bot

Bases: FileServiceMixin, LlmServiceMixin, HttpServiceMixin, IntelServiceMixin

Source code in dandy/bot/bot.py

def __init__(
    self,
    llm_config: str | None = None,
    llm_temperature: float | None = None,
    **kwargs,
) -> None:
    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

def __init_subclass__(cls) -> None:
    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)
                setattr(wrapped, '_wrapped', True)

                return wrapped

            return attr

        cls.__getattribute__ = __getattribute__

__post_init__

Source code in dandy/bot/bot.py

def __post_init__(self) -> None:  # noqa: B027
    pass

get_description classmethod

Source code in dandy/bot/bot.py

@classmethod
def get_description(cls) -> str | None:
    pass

process

Source code in dandy/bot/bot.py

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

def process_to_future(self, *args, **kwargs) -> AsyncFuture:
    return process_to_future(self.process, *args, **kwargs)

reset

Source code in dandy/bot/bot.py

def reset(self) -> None:
    super().reset()

MemoryCache

Bases: BaseCache

cache_name instance-attribute

limit instance-attribute

__len__

Source code in dandy/cache/memory/cache.py

def __len__(self) -> int:
    return len(self._cache)

get

Source code in dandy/cache/memory/cache.py

def get(self, key: str) -> Any | None:
    return self._cache.get(key)

set

Source code in dandy/cache/memory/cache.py

def set(self, key: str, value: Any) -> None:
    self._cache[key] = value
    self.clean()

clean

Source code in dandy/cache/memory/cache.py

def clean(self) -> None:
    if len(self._cache) > self.limit:
        self._cache.popitem(last=False)

clear classmethod

Source code in dandy/cache/memory/cache.py

@classmethod
def clear(cls, cache_name: str = CACHE_DEFAULT_NAME) -> None:
    if cache_name in _memory_cache:
        _memory_cache[cache_name].clear()

clear_all classmethod

Source code in dandy/cache/memory/cache.py

@classmethod
def clear_all(cls) -> None:
    _memory_cache.clear()

destroy_all classmethod

Source code in dandy/cache/memory/cache.py

@classmethod
def destroy_all(cls) -> None:
    cls.clear_all()

SqliteCache

Bases: BaseCache

cache_name instance-attribute

limit instance-attribute

model_post_init

Source code in dandy/cache/sqlite/cache.py

def model_post_init(self, __context: Any, /):
    self._create_table()

__len__

Source code in dandy/cache/sqlite/cache.py

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

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

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

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

@classmethod
def clear(cls, cache_name: str = 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

@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

@classmethod
def destroy_all(cls):
    SqliteConnection(SQLITE_CACHE_DB_NAME).delete_db_file()

DandyCriticalError

Bases: DandyError

DandyError

Bases: Exception

DandyRecoverableError

Bases: DandyError

BaseIntel

Bases: BaseModel, ABC

create_from_file classmethod

Source code in dandy/intel/intel.py

@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

def model_to_kwargs(self) -> dict:
    return dict(self)

model_inc_ex_class_copy classmethod

Source code in dandy/intel/intel.py

@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

@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

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

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

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

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

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

def __getitem__(self, index: int) -> list[T] | T:
    return getattr(self, self._list_name)[index]

__iter__

Source code in dandy/intel/intel.py

def __iter__(self) -> Iterator[T]:
    yield from getattr(self, self._list_name)

__len__

Source code in dandy/intel/intel.py

def __len__(self) -> int:
    return len(getattr(self, self._list_name))

__setitem__

Source code in dandy/intel/intel.py

def __setitem__(self, index: int, value: T):
    getattr(self, self._list_name)[index] = value

append

Source code in dandy/intel/intel.py

def append(self, item: T):
    getattr(self, self._list_name).append(item)

extend

Source code in dandy/intel/intel.py

def extend(self, items: list[T]):
    getattr(self, self._list_name).extend(items)

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

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

def __str__(self) -> str:
    return self.to_str()

to_str

Source code in dandy/llm/prompt/prompt.py

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

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

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

def divider(self) -> Self:
    self.snippets.append(snippet.DividerSnippet())

    return self

array

Source code in dandy/llm/prompt/prompt.py

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

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

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_inner_label=triple_backtick_label
        )
    )

    return self

heading

Source code in dandy/llm/prompt/prompt.py

def heading(
        self,
        heading: str,
) -> Self:
    self.snippets.append(
        snippet.HeadingSnippet(
            heading=heading,
        )
    )

    return self

lb

Source code in dandy/llm/prompt/prompt.py

def lb(self) -> Self:
    return self.line_break()

line_break

Source code in dandy/llm/prompt/prompt.py

def line_break(self) -> Self:
    self.snippets.append(snippet.LineBreakSnippet())

    return self

list

Source code in dandy/llm/prompt/prompt.py

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

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

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

def module_source(
        self,
        module_name: str,
        triple_backtick: bool = True,
        language: str = 'python',
) -> Self:
    self.snippets.append(
        snippet.ModuleSourceSnippet(
            module_name=module_name,
            triple_backtick=triple_backtick,
            triple_backtick_inner_label=language,
            triple_backtick_outer_label=f'module: {module_name}'
        )
    )

    return self

object_source

Source code in dandy/llm/prompt/prompt.py

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_inner_label=object_module_name,
        )
    )

    return self

ordered_list

Source code in dandy/llm/prompt/prompt.py

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

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

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

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

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_inner_label=triple_backtick_label,
        )
    )

    return self

title

Source code in dandy/llm/prompt/prompt.py

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

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

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

@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

@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

@classmethod
def delete_all_recordings(cls):
    cls.recordings.clear()

delete_recording classmethod

Source code in dandy/recorder/recorder.py

@classmethod
def delete_recording(cls, recording_name: str = RECORDING_DEFAULT_NAME):
    cls.check_recording_is_valid(recording_name)
    del cls.recordings[recording_name]

get_default_recording_path classmethod

Source code in dandy/recorder/recorder.py

@classmethod
def get_default_recording_path(cls) -> Path:
    return Path(settings.BASE_PATH, settings.DANDY_DIRECTORY, RECORDING_OUTPUT_DIRECTORY)

get_recording classmethod

Source code in dandy/recorder/recorder.py

@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

@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

@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

@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

@classmethod
def stop_all_recording(cls):
    for recording in cls.recordings.values():
        recording.stop()

to_file classmethod

Source code in dandy/recorder/recorder.py

@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

@classmethod
def to_html_file(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
        path: Path | str | None = None
):
    if path is None:
        path = cls.get_default_recording_path()

    cls.to_file(
        recording_name,
        'html',
        path
    )

to_html_str classmethod

Source code in dandy/recorder/recorder.py

@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

@classmethod
def to_json_file(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
        path: Path | str | None = None
):
    if path is None:
        path = cls.get_default_recording_path()

    cls.to_file(
        recording_name,
        'json',
        path
    )

to_json_str classmethod

Source code in dandy/recorder/recorder.py

@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

@classmethod
def to_markdown_file(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
        path: Path | str | None = None
):
    if path is None:
        path = cls.get_default_recording_path()

    cls.to_file(
        recording_name,
        'markdown',
        path
    )

to_markdown_str classmethod

Source code in dandy/recorder/recorder.py

@classmethod
def to_markdown_str(
        cls,
        recording_name: str = RECORDING_DEFAULT_NAME,
) -> str:
    return cls._to_str(
        recording_name,
        'markdown',
    )

cache_to_memory

Source code in dandy/cache/memory/decorators.py

def cache_to_memory(
        cache_name: str = CACHE_DEFAULT_NAME,
        limit: int | None = None,
) -> Callable:
    if limit is None:
        limit = settings.CACHE_MEMORY_LIMIT

    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

def cache_to_sqlite(
        cache_name: str = CACHE_DEFAULT_NAME,
        limit: int | None = None
) -> Callable:
    if limit is None:
        limit = settings.CACHE_SQLITE_LIMIT

    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

generate_cache_key

Source code in dandy/cache/tools.py

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

    return hashlib.shake_128(
        str(hashable_tuple).encode()
    ).hexdigest(16)

process_to_future

Source code in dandy/core/future/tools.py

def process_to_future(
        callable_: Callable[..., R],
        *args,
        **kwargs
) -> AsyncFuture[R]:
    return AsyncFuture[R](callable_, *args, **kwargs)

recorder_to_html_file

Source code in dandy/recorder/decorators.py

def recorder_to_html_file(recording_name: str | None = None, path: Path | str | None = None) -> Callable:
    if path is None:
        path = Recorder.get_default_recording_path()

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Callable:
            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

def recorder_to_json_file(recording_name: str | None = None, path: Path | str | None = None) -> Callable:
    if path is None:
        path = Recorder.get_default_recording_path()

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Callable:
            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

def recorder_to_markdown_file(recording_name: str | None = None, path: Path | str | None = None) -> Callable:
    if path is None:
        path = Recorder.get_default_recording_path()

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Callable:
            return _recorder_to_file_decorator_function(func, args, kwargs, recording_name, 'markdown', path)

        return wrapper

    return decorator