Utilities¶

Public API for kstlib.utils, a collection of utility helpers shared across kstlib modules. These functions handle common tasks like dictionary merging, value formatting, lazy initialization, secure file deletion, and email validation.

Tip

These utilities are building blocks used internally by other kstlib modules but are also available for direct use in your applications.

Quick overview¶

Dictionary helpers:

deep_merge(base, override) - Recursively merge dictionaries with override semantics

Formatting:

format_bytes(size) - Human-readable byte sizes (e.g., “1.5 GB”)
format_count(n) - Human-readable counts with SI suffixes
format_duration(seconds) - Duration as “Xh Ym Zs”
format_time_delta(delta) - Format timedelta objects
format_timestamp(epoch, fmt, tz) - Config-driven epoch to datetime string
parse_size_string(s) - Parse “10MB” to bytes

Lazy initialization:

lazy_factory(factory) - Decorator for lazy singleton initialization

Secure deletion:

secure_delete(path) - Securely delete files with configurable overwrite methods
SecureDeleteMethod - Enum of deletion strategies (zeros, random, DoD, Gutmann)
SecureDeleteReport - Result object with deletion details

Text processing:

replace_placeholders(template, values) - Simple {key} placeholder substitution

Validators:

parse_email_address(s) - Parse email string to EmailAddress object
normalize_address_list(addresses) - Normalize list of email addresses
EmailAddress - Named tuple for parsed email components
ValidationError - Raised on invalid input

Dictionary Helpers¶

deep_merge¶

kstlib.utils.deep_merge(base: 'dict[str, Any]', updates: 'Mapping[str, Any]', *, deep_copy: 'bool' = False, _depth: 'int' = 0) → 'dict[str, Any]' -> dict[str, Any][source]

Recursively merge updates into base dictionary (in place).

Parameters:

base (dict[str, Any]) – Base dictionary to update (modified in place).
updates (Mapping[str, Any]) – Dictionary with updates to merge.
deep_copy (bool) – If True, deep copy values before assignment.
_depth (int) – Internal recursion counter (do not set manually).

Returns:

The modified base dictionary (for chaining).

Raises:

RecursionError – If nesting exceeds MAX_MERGE_DEPTH.

Return type:

dict[str, Any]

Examples

>>> base = {"a": {"x": 1}, "b": 2}
>>> deep_merge(base, {"a": {"y": 2}, "c": 3})
{'a': {'x': 1, 'y': 2}, 'b': 2, 'c': 3}

Formatting Functions¶

format_bytes¶

kstlib.utils.format_bytes(size: 'float', binary: 'bool' = True) → 'str' -> str[source]

Format a byte count as a human-readable string.

Parameters:

size (float) – Size in bytes (int or float).
binary (bool) – If True, use binary units (KiB, MiB). If False, use SI units (KB, MB).

Returns:

Human-readable size string (e.g., “25.0 MiB” or “25.0 MB”).

Return type:

str

Examples

>>> format_bytes(25 * 1024 * 1024)
'25.0 MiB'
>>> format_bytes(25 * 1000 * 1000, binary=False)
'25.0 MB'

format_count¶

kstlib.utils.format_count(value: 'int') → 'str' -> str[source]

Format a count with comma separators for readability.

Parameters:: value (int) – Integer count to format.
Returns:: Comma-separated string (e.g., “1,000,000”).
Return type:: str

Examples

>>> format_count(1000000)
'1,000,000'

format_duration¶

kstlib.utils.format_duration(seconds: 'float') → 'str' -> str[source]

Format a duration in seconds as a human-readable string.

Parameters:: seconds (float) – Duration in seconds.
Returns:: Human-readable duration (e.g., “5 minutes”, “2 hours”).
Return type:: str

Examples

>>> format_duration(300)
'5 minutes'
>>> format_duration(3661)
'an hour'

format_time_delta¶

kstlib.utils.format_time_delta(dt: 'datetime', other: 'datetime | None' = None) → 'str' -> str[source]

Format a datetime as a relative time string.

Parameters:

dt (datetime) – Target datetime.
other (datetime | None) – Reference datetime (defaults to now).

Returns:

Relative time string (e.g., “2 hours ago”, “in 3 days”).

Return type:

str

Examples

>>> from datetime import datetime, timedelta
>>> past = datetime.now() - timedelta(hours=2)
>>> format_time_delta(past)
'2 hours ago'

format_timestamp¶

kstlib.utils.format_timestamp(epoch: 'float | str | None', fmt: 'str | None' = None, tz: 'str | None' = None) → 'str' -> str[source]

Format an epoch timestamp as a human-readable datetime string.

Converts Unix epoch timestamps to formatted datetime strings using pendulum for timezone-aware formatting. Configuration can be loaded from kstlib.conf.yml or provided explicitly.

Parameters:

epoch (float | str | None) – Unix timestamp (seconds since 1970-01-01 UTC). Accepts int, float, or string representation. Returns “(invalid)” if None or unparseable.
fmt (str | None) – Datetime format string (pendulum tokens). If None, uses config value or “YYYY-MM-DD HH:mm:ss”.
tz (str | None) – Timezone for display (“local”, “UTC”, or IANA name). If None, uses config value or “local”.

Returns:

Formatted datetime string, or “(invalid)” on error.

Return type:

str

Examples

>>> format_timestamp(1706234567, tz="UTC")
'2024-01-26 02:02:47'
>>> format_timestamp(1706234567, fmt="DD/MM/YYYY", tz="UTC")
'26/01/2024'
>>> format_timestamp(None)
'(invalid)'

parse_size_string¶

kstlib.utils.parse_size_string(value: 'str | float') → 'int' -> int[source]

Parse a human-readable size string into bytes.

Accepts raw integers, floats, or strings with optional units. Supported units: B, K, KB, KiB, M, MB, MiB, G, GB, GiB, T, TB, TiB.

Parameters:: value (str | float) – Size as int, float, or string with optional unit suffix.
Returns:: Size in bytes as an integer.
Raises:: ValueError – If the string format is invalid or the unit is unknown.
Return type:: int

Examples

>>> parse_size_string(1024)
1024
>>> parse_size_string("25M")
26214400
>>> parse_size_string("100 MiB")
104857600
>>> parse_size_string("1.5GB")
1610612736

Lazy Initialization¶

lazy_factory¶

kstlib.utils.lazy_factory(module_path: 'str', class_name: 'str') → 'Callable[[Callable[..., T]], Callable[..., T]]' -> Callable[[Callable[..., T]], Callable[..., T]][source]

Defer import of a class until the factory function is first called.

Defers the import of the specified class until the factory is actually called, reducing startup time when the factory is registered but not used.

Parameters:

module_path (str) – Full dotted path to the module containing the class.
class_name (str) – Name of the class to import from the module.

Returns:

A decorator that wraps the factory function with lazy import behavior.

Return type:

Callable[[Callable[…, T]], Callable[…, T]]

Example

>>> @lazy_factory("kstlib.secrets.providers.sops", "SOPSProvider")
... def _sops_factory(**kwargs):
...     ...  # Body is ignored, class is instantiated automatically
>>>
>>> # SOPSProvider is only imported when _sops_factory() is called
>>> provider = _sops_factory(path="secrets.sops.yml")

Secure Deletion¶

secure_delete¶

kstlib.utils.secure_delete(target: 'Path | str', *, passes: 'int' = 3, method: 'SecureDeleteMethod | str' = <SecureDeleteMethod.AUTO: 'auto'>, chunk_size: 'int' = 1048576, zero_last_pass: 'bool' = True) → 'SecureDeleteReport' -> SecureDeleteReport[source]

Securely remove target from disk.

Parameters:

target (Path | str) – File path that must be removed.
passes (int) – Number of overwrite passes to perform when relying on the built-in overwrite implementation. Values lower than 1 raise ValueError.
method (SecureDeleteMethod | str) – Preferred strategy. auto attempts to use a platform shred command and falls back to overwriting when none is available or when the command fails. command forces the usage of a system command and reports an error if it is not available. overwrite forces the Python overwrite implementation.
chunk_size (int) – Size, in bytes, of the chunks written during the overwrite loop. Defaults to 1 MiB.
zero_last_pass (bool) – If True, the final overwrite pass writes zeros instead of random data.

Returns:

A SecureDeleteReport describing the outcome.

Raises:

ValueError – If passes is lower than 1 or if target does not reference a regular file.

Return type:

SecureDeleteReport

Example

Securely remove a cleartext file once it is no longer needed:

>>> from pathlib import Path
>>> from kstlib.utils.secure_delete import secure_delete, SecureDeleteMethod
>>> path = Path("secret.txt")
>>> _ = path.write_text("classified")  
>>> report = secure_delete(path, method=SecureDeleteMethod.OVERWRITE, passes=1)  
>>> report.success  
True

SecureDeleteMethod¶

class kstlib.utils.SecureDeleteMethod(value)[source]

Bases: str, Enum

Available strategies for securely deleting files.

AUTO = 'auto'

COMMAND = 'command'

OVERWRITE = 'overwrite'

__format__(self, format_spec): Returns format using actual value type unless __str__ has been overridden.

SecureDeleteReport¶

class kstlib.utils.SecureDeleteReport(success, method, passes, command=None, message=None)[source]

Bases: object

Summary result produced by secure_delete().

success: bool

method: SecureDeleteMethod

passes: int

command: Sequence[str] | None

message: str | None

__init__(self, success: 'bool', method: 'SecureDeleteMethod', passes: 'int', command: 'Sequence[str] | None' = None, message: 'str | None' = None) → None -> None

Text Processing¶

replace_placeholders¶

kstlib.utils.replace_placeholders(template: 'str', values: 'Mapping[str, Any] | None' = None, /, **kwargs: 'Any') → 'str' -> str[source]

Replace {{ placeholder }} tokens within template.

Parameters:

template (str) – Raw template string containing placeholder patterns.
values (Mapping[str, Any] | None) – Optional mapping used to look up replacement values. When provided, it is merged with the keyword arguments, giving precedence to the latter.
**kwargs (Any) – Additional placeholder values.

Returns:

Rendered template with matching placeholders substituted by their string representation. Missing placeholders are left untouched to simplify incremental rendering.

Return type:

str

Examples

>>> replace_placeholders("Hello {{ name }}!", name="Ada")
'Hello Ada!'

Validators¶

parse_email_address¶

kstlib.utils.parse_email_address(value: 'str') → 'EmailAddress' -> EmailAddress[source]

Parse value into a validated EmailAddress.

Parameters:: value (str) – Raw email string, optionally containing a display name.
Returns:: A normalized EmailAddress instance.
Raises:: ValidationError – If the string does not contain a valid address.
Return type:: EmailAddress

Examples

>>> parse_email_address("Ada Lovelace <ADA@example.COM>").formatted
'Ada Lovelace <ada@example.com>'
>>> parse_email_address("foo@bar")
Traceback (most recent call last):
...
kstlib.utils.validators.ValidationError: Invalid email address: 'foo@bar'

normalize_address_list¶

kstlib.utils.normalize_address_list(values: 'Iterable[str]') → 'list[EmailAddress]' -> list[EmailAddress][source]

Validate and normalize a sequence of email addresses.

Examples

>>> normalize_address_list([
...     "Ada Lovelace <ada@example.com>",
...     "grace@example.net",
... ])  
[EmailAddress(name='Ada Lovelace', address='ada@example.com'),
 EmailAddress(name='', address='grace@example.net')]

EmailAddress¶

class kstlib.utils.EmailAddress(name, address)[source]

Bases: object

Normalized representation of an email address.

name: str

address: str

property formatted: str: Return "Name <email@domain>" if a display name is present.

__init__(self, name: 'str', address: 'str') → None -> None

ValidationError¶

class kstlib.utils.ValidationError[source]

Bases: KstlibError, ValueError

Raised when user supplied values fail validation.

HTTP Tracing¶

HTTPTraceLogger¶

class kstlib.utils.HTTPTraceLogger(logger, *, trace_level=5, sensitive_keys=None, pretty_print=True, max_body_length=2000)[source]

Bases: object

Reusable HTTP trace logger with sensitive data redaction.

This class provides httpx event hooks for logging HTTP requests and responses at TRACE level with automatic redaction of sensitive data.

Parameters:

logger (logging.Logger) – Logger instance to use for trace output.
trace_level (int) – Logging level for trace messages (default: 5 for TRACE).
sensitive_keys (frozenset[str] | None) – Set of keys to redact in request bodies.
pretty_print (bool) – Whether to pretty-print JSON responses.
max_body_length (int) – Maximum response body length before truncation.

Examples

>>> import logging
>>> import httpx
>>> from kstlib.utils.http_trace import HTTPTraceLogger
>>> tracer = HTTPTraceLogger(logging.getLogger(__name__))
>>> client = httpx.Client(
...     event_hooks={
...         "request": [tracer.on_request],
...         "response": [tracer.on_response],
...     }
... )

__init__(self, logger: 'logging.Logger', *, trace_level: 'int' = 5, sensitive_keys: 'frozenset[str] | None' = None, pretty_print: 'bool' = True, max_body_length: 'int' = 2000) → 'None' -> None[source]

Initialize the HTTP trace logger.

property sensitive_keys: frozenset[str]: Return the set of sensitive keys being redacted.

configure(self, *, pretty_print: 'bool | None' = None, max_body_length: 'int | None' = None) → 'None' -> None[source]

Update trace configuration at runtime.

Parameters:

pretty_print (bool | None) – Whether to pretty-print JSON responses.
max_body_length (int | None) – Maximum response body length before truncation.

on_request(self, request: 'httpx.Request') → 'None' -> None[source]

Httpx event hook for outgoing requests (TRACE logging).

Redacts sensitive data in request body and Authorization headers.

Parameters:: request (httpx.Request) – The outgoing HTTP request.

on_response(self, response: 'httpx.Response') → 'None' -> None[source]

Httpx event hook for incoming responses (TRACE logging).

Optionally pretty-prints JSON and truncates long bodies.

Parameters:: response (httpx.Response) – The incoming HTTP response.