Raccoon Tools

This is a collection of tools that I regularly use on several projects. To stop duplicating and to (hopefully) help someone, I decided to make them into a public package.

Functionalities

Decorators

retry

A decorator that retries a function call a specified number of times before giving up. It logs each attempt and the final failure if all retries are exhausted.

Parameters:

• retries: Maximum number of retries before giving up (default: 3).
• delay: Delay in seconds between each retry (default: 1).
• delay_is_exponential: If True, the delay between retries will increase exponentially (default: False).
• only_exceptions_of_type: A list of exception types to catch and retry on. If None, all exceptions are caught.
• log_level: The log level used by the decorator (default: logging.ERROR).

Example:

from raccoontools.decorators.retry import retry

attempts = {"count": 0}

@retry(retries=4, delay=0.25)
def flaky_operation():
    attempts["count"] += 1
    if attempts["count"] < 3:
        raise RuntimeError("Still failing...")
    return "Success!"

print(flaky_operation())

retry_request

Like the previous decorator, but for HTTP requests. It logs each attempt and the final failure if all retries are exhausted. It also provides options to handle specific HTTP status codes.

Parameters:

• retries: Maximum number of retries before giving up (default: 3).
• delay: Delay in seconds between each retry (default: 1).
• delay_is_exponential: If True, the delay between retries will increase exponentially (default: False).
• skip_retry_on_404: If True, the decorator will not retry on 404 responses (default: False).
• retry_only_on_status_codes: A list of HTTP status codes to retry on. If None, no retries will be made.
• get_new_token_on_401: An optional callable to execute and get a new token when a 401 response is received.
• get_new_token_on_403: An optional callable to execute and get a new token when a 403 response is received.
• log_level: The log level used by the decorator (default: logging.ERROR).

Example:

import requests
from raccoontools.decorators.retry import retry_request

@retry_request(retries=5, delay=1, skip_retry_on_404=True)
def call_service():
    return requests.get("https://httpbin.org/status/500")

response = call_service()
print(response.status_code)

benchmark

A decorator that benchmarks the execution time of a function. The results are logged using the logging module at the INFO level. The decorated function can also provide benchmark information via the get_benchmark_info method.

Example:

import time
from raccoontools.decorators.benchmark import benchmark

@benchmark
def slow_function():
    time.sleep(0.5)
    return "done"

slow_function()
print(slow_function.get_benchmark_info())

Generators

infinite_iterator

Generates an infinite iterator from a list.

Parameters:

• list_to_iterate_over: The list to iterate over.

Example:

from raccoontools.generators.misc_generators import infinite_iterator

rotating_status = infinite_iterator(["⏳", "⌛", "✅"])
for _ in range(6):
    print(next(rotating_status), end=" ")

read_line

Reads a file line by line.

Parameters:

• file: Path to the file.
• strip_line: Strip whitespace from the beginning and end of each line (default: True).
• encoding: File encoding (default: 'utf-8').
• buffer_size: Size of the read buffer in bytes. If None, the default system buffer is used.

Example:

from pathlib import Path
from raccoontools.generators.file_ops_generators import read_line

for line in read_line(Path("logs/app.log"), strip_line=True):
    print(line)

read_csv

Reads a CSV file line by line, yielding each line as a dictionary + metadata.

Parameters:

• file: Path to the CSV file.
• encoding: File encoding (default: 'utf-8').
• has_headers: If True, the first line of the CSV file is treated as a header.
• buffer_size: Size of the read buffer in bytes. If None, the default system buffer is used.

Example:

from raccoontools.generators.file_ops_generators import read_csv

for row, metadata in read_csv("data/report.csv", has_headers=True):
    print(metadata.index, row)

id_guid_generator

Generates unique GUID (Globally Unique Identifier) strings.

Parameters:

• ids_to_generate: The number of GUIDs to generate. If None, generates indefinitely.

Example:

from raccoontools.generators.sequence_generators import id_guid_generator

guid_gen = id_guid_generator(ids_to_generate=2)
print(list(guid_gen))

id_int_generator

Generates integer IDs with optional validation.

Parameters:

• ids_to_generate: The number of IDs to generate. If None, generates indefinitely.
• start_at: The starting value for the ID sequence (default: 0).
• validate_id: A function to validate each ID. If None, all IDs are considered valid.

Example:

from raccoontools.generators.sequence_generators import id_int_generator

only_even_ids = id_int_generator(ids_to_generate=3, start_at=10, validate_id=lambda x: x % 2 == 0)
print(list(only_even_ids))  # [10, 12, 14]

timestamp_generator

Generates Unix timestamps.

Parameters:

• timestamps_to_generate: The number of timestamps to generate. If None, generates indefinitely.

Example:

from raccoontools.generators.sequence_generators import timestamp_generator

timestamps = []
for ts in timestamp_generator(3):
    timestamps.append(ts)
print(timestamps)

sentence_generator

Generates Lorem Ipsum sentences with lengths ranging from min_length to max_length.

Parameters:

• sentences_to_generate: The number of sentences to generate. If None, generates indefinitely.
• min_length: The minimum length of each sentence (default: 1).
• max_length: The maximum length of each sentence. If None, a random value between 10 and 512 is used for each sentence.

Example:

from raccoontools.generators.sequence_generators import sentence_generator

for sentence in sentence_generator(2, min_length=40, max_length=80):
    print(sentence)

Converters

timestamp_to_datetime

Converts a numeric timestamp to a timezone-aware datetime object. Uses integer arithmetic for int inputs to preserve full microsecond precision without IEEE-754 float rounding.

Parameters:

• timestamp: The numeric timestamp value (int or float).
• unit: Unit of the input: "s", "ms", "us", or "ns" (default: "ms").
• tz: Target timezone (default: timezone.utc). Pass None to get a naive datetime.

Example:

from raccoontools.converters.datetime_converters import timestamp_to_datetime

# Milliseconds (default) — e.g. Nightscout, JavaScript Date.now()
dt = timestamp_to_datetime(1_700_000_000_000)
print(dt)  # 2023-11-14 22:13:20+00:00

# Seconds
dt = timestamp_to_datetime(1_700_000_000, unit="s")

# Nanoseconds — sub-microsecond part is truncated
dt = timestamp_to_datetime(1_700_000_000_123_456_789, unit="ns")
print(dt.microsecond)  # 123456

# Custom timezone
from datetime import timezone, timedelta
tz_ist = timezone(timedelta(hours=5, minutes=30))
dt = timestamp_to_datetime(1_700_000_000_000, tz=tz_ist)
print(dt)  # 2023-11-15 03:43:20+05:30

timedelta_to_readable_elapsed_time

Converts a timedelta to a human-readable elapsed time string (e.g. "2 days and 5 hours ago"). Uses approximate 30-day months and 365-day years.

Parameters:

• delta: The timedelta to format. Must be non-negative.
• suffix: Text appended after the time string (default: "ago"). Pass "" or None to omit.
• zero_delta_label: Returned when the delta is below zero_delta_threshold (default: "just now").
• zero_delta_threshold: Deltas below this value return zero_delta_label (default: timedelta(seconds=60)). Pass None to always show the real representation (e.g. "30 seconds ago").

Example:

from datetime import timedelta
from raccoontools.converters.datetime_converters import timedelta_to_readable_elapsed_time

print(timedelta_to_readable_elapsed_time(timedelta(days=400)))
# '1 year, 1 month and 5 days ago'

print(timedelta_to_readable_elapsed_time(timedelta(hours=1, minutes=30)))
# '1 hour and 30 minutes ago'

print(timedelta_to_readable_elapsed_time(timedelta(seconds=10)))
# 'just now'

# Show seconds instead of "just now"
print(timedelta_to_readable_elapsed_time(timedelta(seconds=30), zero_delta_threshold=None))
# '30 seconds ago'

print(timedelta_to_readable_elapsed_time(timedelta(hours=2), suffix="elapsed"))
# '2 hours elapsed'

print(timedelta_to_readable_elapsed_time(timedelta(hours=2), suffix=None))
# '2 hours'

time_value_to_readable_elapsed_time

Convenience wrapper that converts a numeric value + unit into a timedelta and delegates to timedelta_to_readable_elapsed_time.

Parameters:

• value: The numeric amount (int or float). Must be non-negative and finite.
• time_piece: Unit — one of "milliseconds", "seconds", "minutes", "hours", "days", "weeks" (default: "seconds").
• suffix: Text appended after the time string (default: "ago").
• zero_delta_label: Returned when the resulting delta is below zero_delta_threshold (default: "just now").
• zero_delta_threshold: Deltas below this value return zero_delta_label (default: timedelta(seconds=60)). Pass None to always show the real representation.

Example:

from raccoontools.converters.datetime_converters import time_value_to_readable_elapsed_time

print(time_value_to_readable_elapsed_time(3600))
# '1 hour ago'

print(time_value_to_readable_elapsed_time(2, time_piece="days"))
# '2 days ago'

print(time_value_to_readable_elapsed_time(1.5, time_piece="hours"))
# '1 hour and 30 minutes ago'

Comparators

dicts_are_equal / compare_dicts

Deeply compares two dictionaries, ignoring key order. Nested dictionaries are recursed into; every other value (nested lists, datetime objects, custom types, ...) is compared as an opaque leaf via ==.

• dicts_are_equal(a: dict, b: dict) -> bool: Returns True if the dictionaries are deeply equal.
• compare_dicts(a: dict, b: dict) -> DictComparison: Returns a named tuple (are_equal, differences). differences is a DictDiff with added / removed / changed maps, keyed by dotted paths (e.g. "user.address.zip"). Both raise TypeError for non-dict input.

Example:

from raccoontools.comparators.dict_comparators import dicts_are_equal, compare_dicts

print(dicts_are_equal({"a": 1, "b": 2}, {"b": 2, "a": 1}))  # True

result = compare_dicts(
    {"name": "Ada", "role": "dev"},
    {"name": "Ada", "role": "lead", "active": True},
)
print(result.are_equal)              # False
print(result.differences.changed)    # {'role': ('dev', 'lead')}
print(result.differences.added)      # {'active': True}

dict_lists_are_equal / compare_dict_lists

Compares two lists of dictionaries ignoring their order. The lists are treated as multisets, so duplicates are counted. Lists nested inside a dictionary remain order-sensitive.

• dict_lists_are_equal(a: list[dict], b: list[dict]) -> bool: Returns True if both lists hold the same dictionaries in any order.
• compare_dict_lists(a: list[dict], b: list[dict]) -> DictListComparison: Returns a named tuple (are_equal, differences). differences is a DictListDiff with only_in_a / only_in_b. Both raise TypeError if an argument is not a list or an element is not a dict.

Example:

from raccoontools.comparators.dict_comparators import dict_lists_are_equal, compare_dict_lists

print(dict_lists_are_equal([{"id": 1}, {"id": 2}], [{"id": 2}, {"id": 1}]))  # True

result = compare_dict_lists([{"id": 1}, {"id": 2}], [{"id": 2}, {"id": 3}])
print(result.are_equal)              # False
print(result.differences.only_in_a)  # [{'id': 1}]
print(result.differences.only_in_b)  # [{'id': 3}]

Shared Utilities

file_ops

Provides functions to load and save JSON data to and from files.

• load_json_from_file(file: Path | str, encoding: str = "utf-8", object_hook: Callable | None = obj_dump_deserializer) -> dict | list[dict]: Loads a JSON file and returns the data as a dictionary or list of dictionaries. Accepts a Path or a string path. By default, uses obj_dump_deserializer to reconstruct types (datetime, int, float, Path). Pass object_hook=None for raw JSON parsing with no type coercion.
• save_json_to_file(data: dict | list[dict], target_file_or_folder: Path | str, dump_kwargs: dict | None = None, encoding: str = "utf-8") -> Path: Saves a dictionary or list of dictionaries to a JSON file. Accepts a Path or a string path (an empty string raises ValueError).

Example:

from pathlib import Path
from raccoontools.shared.file_ops import load_json_from_file, save_json_to_file

payload = {"status": "ok"}
target_dir = Path("artifacts")
target_dir.mkdir(exist_ok=True)
saved_file = save_json_to_file(payload, target_dir)
loaded_payload = load_json_from_file(saved_file)
print(saved_file, loaded_payload)

file_utils

Provides utility functions for file operations.

• get_filename_for_new_file(file_extension: str, prefix: str | None = None, add_current_datetime_as_format: str = "%Y%m%d%H%M%S%f", use_utc: bool = True, unique_identifier: str | bool = True, part_separator: str = "-", suffix: str | None = None) -> str: Generates a unique filename for a new file.
• get_date_based_subfolder(ref_path: Path | str, use_utc: bool = True, date_ref: datetime | None = None, add_delta_days: int | None = None, date_format: str = "%Y-%m-%d", create_if_missing: bool = True) -> Path: Gets (or creates) a date-based subfolder under the given path. Accepts a Path or a string path. If ref_path points to a file (or has a file extension), uses its parent directory as the base.

Example:

from pathlib import Path
from raccoontools.shared.file_utils import get_filename_for_new_file, get_date_based_subfolder

filename = get_filename_for_new_file("json", prefix="snapshot", suffix="v1")
print(filename)  # snapshot-20240201130501999999-...-v1.json

folder = get_date_based_subfolder(Path("output/data"), add_delta_days=-1)
print(folder)  # output/data/2025-03-31

string_utils

Provides utility functions for string manipulation.

• linefy(text: str) -> str: Collapses a multi-line string into a single line. Removes all linebreaks, merges multiple sequential whitespace characters (spaces, tabs, newlines, etc.) into a single space, and trims leading and trailing whitespace. Raises TypeError if text is not a str.

Example:

from raccoontools.shared.string_utils import linefy

messy = "  Hello\n\tworld  \n  again  "
print(linefy(messy))  # Hello world again

http

Provides utility functions for HTTP headers.

• get_headers(token: str | None = None, content_type: str = "application/json", user_agent: str | None = None, fake_browser_user_agent: bool = False, extra_args: dict[str, str] | None = None) -> dict[str, str]: Generates headers for an HTTP request.

Example:

from raccoontools.shared.http import get_headers

headers = get_headers(
    token="abc123",
    fake_browser_user_agent=True,
    extra_args={"X-Trace-Id": "req-1"}
)
print(headers)

requests_with_retry

A wrapper around requests using the retry_request decorator. Can be used as a drop-in replacement for requests.

• get(url, params=None, **kwargs) -> requests.Response: Sends a GET request with retry functionality.
• options(url, **kwargs) -> requests.Response: Sends an OPTIONS request with retry functionality.
• head(url, **kwargs) -> requests.Response: Sends a HEAD request with retry functionality.
• post(url, data=None, json=None, **kwargs) -> requests.Response: Sends a POST request with retry functionality.
• put(url, data=None, **kwargs) -> requests.Response: Sends a PUT request with retry functionality.
• patch(url, data=None, **kwargs) -> requests.Response: Sends a PATCH request with retry functionality.
• delete(url, **kwargs) -> requests.Response: Sends a DELETE request with retry functionality.

Example:

import raccoontools.shared.requests_with_retry as requests

response = requests.get("https://httpbin.org/status/500")
print(response.status_code)

serializer

Provides functions to serialize and deserialize objects.

• serialize_to_dict(obj) -> dict | list[dict] | None: Serializes an object to a dictionary or list of dictionaries.
• parse_csv(csv_data: str) -> list[dict]: Parses a CSV string and returns a list of dictionaries.
• csv_string_to_dict_list(data: str | list[str] | dict | list[dict], no_data_return: str = "No data available") -> list[dict] | str: Converts a CSV string to a list of dictionaries.
• dataset_to_prompt_text(dataset: list[dict]) -> str: Converts a dataset to a prompt text.
• obj_dump_serializer(obj): Serializes objects for saving data to a file.
• obj_dump_deserializer(obj): Deserializes objects when loading data from a file.

Example:

from pydantic import BaseModel
from raccoontools.shared.serializer import serialize_to_dict, csv_string_to_dict_list

class User(BaseModel):
    name: str
    active: bool

payload = serialize_to_dict(User(name="Ada", active=True))
print(payload)  # {'name': 'Ada', 'active': True}

rows = csv_string_to_dict_list("name,score\nAda,10\nBob,7\n")
print(rows)

Changelog

• Check the changelog here.