edri.utility package

Subpackages

• edri.utility.manager package
  • Submodules
  • edri.utility.manager.scheduler module
    • Job
      • Job.repeat
  • edri.utility.manager.store module
    • Callback
  • Module contents

Submodules

edri.utility.cache module

class edri.utility.cache.Cache

Bases: object

last_change(key: str) → datetime

renew(key: str) → None

seed(key: str, when: datetime) → None

Set the last-change timestamp for key explicitly.

Use from Manager.after_start() to pre-seed a cache key so its first ETag reflects a known baseline instead of process start time. A naive when is assumed to be in the cache timezone, keeping it comparable with the aware timestamps renew() produces.

tag(key: str) → str

edri.utility.coercion module

Process-agnostic value coercion.

Turns a raw incoming value into the type declared by a dataclass field's annotation, running any Injection validators/transformers along the way. This logic is intentionally free of any request/transport state so it can be reused outside the API Listener — for example by the Broker when coercing WebSocket subscription filter values to the target event field's type.

The behaviour lives on TypeCoercer. The API handlers inherit it (so an http_handler can still override TypeCoercer.convert_type()), while non-handler components instantiate it directly.

class edri.utility.coercion.TypeCoercer

Bases: object

Converts raw values to a declared annotation type.

Stateless — every method depends only on its arguments, so a single shared instance is safe to reuse across calls and threads.

convert_type(value: Any, annotation: type) → Any

Validates and converts input values to the specified annotation type,

supporting:

• basic types

• Optional / Union / |

• list / tuple / dict with type args

• Literal

• list-like subclasses (e.g. ListValidation[int])

• Injection of validation classes (e.g. Injection((ListValidation[int],), {...}))

• Nullable injection via inject(...) | None

edri.utility.function module

edri.utility.function.camel2snake(title: str) → str

Converts a camelCase or PascalCase string to a snake_case string.

Parameters:

title (str) -- The camelCase or PascalCase string to be converted.

Returns:

The converted snake_case string.

Return type:

str

Examples:

>>> camel2snake("camelCaseExample")
'camel_case_example'
>>> camel2snake("CamelCaseExample")
'camel_case_example'

edri.utility.function.inject(*args: Callable[[...], Any], **kwargs: Any) → Injection

Creates an Injection for validator callables and their parameters.

This function is typically used to declaratively define dependency injections by specifying a class and any keyword arguments to pass to its constructor.

Parameters:

• *args -- Validator classes or functions called in order.

• **kwargs -- Keyword arguments passed to validators whose signatures accept them.

Returns:

An Injection object wrapping the class and its parameters.

Return type:

Injection[T]

Example

inject(MyService, config=..., db=...)

Use | None to make the injection optional:

inject(StringValidator, minimum_length=2) | None

edri.utility.function.snake2camel(title: str, /, *, exceptions: tuple[str, ...] | None = None) → str

Converts a snake_case string to a camelCase string, with optional exceptions where the conversion should not be applied.

Parameters:

• title (str) -- The snake_case string to be converted.

• exceptions (Optional[Tuple[str, ...]]) -- A tuple of strings that, when matched at the beginning of the title, prevent conversion and return the title as is. Defaults to None.

Returns:

The converted camelCase string if no exception is matched; otherwise, the original string.

Return type:

str

Raises:

KeyError -- If a non-trailing URL parameter is missing and subsequent placeholders are present.

Examples:

>>> snake2camel("snake_case_example")
'snakeCaseExample'
>>> snake2camel("id_example", exceptions=("id_",))
'id_example'

edri.utility.json_encoder module

class edri.utility.json_encoder.CustomJSONEncoder(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, sort_keys: bool = False, indent: int | str | None = None, separators: tuple[str, str] | None = None, default: Callable[[Any], Any] | None = None, context: dict[str, Any] | None = None)

Bases: JSONEncoder

Custom JSON encoder that extends Python's built-in JSONEncoder to handle additional Python types and custom serialization logic.

Supported types and custom behaviors: - Objects with a to_json() method: Serialized using that method. - Datetime-like objects with isoformat(): Serialized to ISO 8601 format. - Path objects: Serialized as POSIX-style strings. - bytes and bytearray: Encoded as hexadecimal strings. - Enum members: Serialized using their .value. - UUID instances: Serialized as hexadecimal strings. - Exception instances: Serialized as a dictionary with type, message, and args. - Dataclass instances: Converted to dictionaries via asdict.

Parameters:

• skipkeys (bool) -- Whether to skip keys not of a basic type (str, int, float, bool, None).

• ensure_ascii (bool) -- If True, escapes all non-ASCII characters in output.

• check_circular (bool) -- If True, checks for circular references and raises an error if found.

• allow_nan (bool) -- If True, allows NaN and Infinity values (JavaScript-compatible).

• sort_keys (bool) -- If True, output dictionary keys are sorted.

• indent (int or None) -- If specified, pretty-prints output with given number of spaces.

• separators (tuple or None) -- Tuple of (item_separator, key_separator) for customizing output.

• default (callable or None) -- A fallback function for unsupported objects.

• context (dict or None) -- Optional context dictionary for passing extra encoding parameters.

default(data: Any) → Any

Serializes unsupported data types into JSON-compatible formats.

Custom serialization behavior: - If object has to_json() method: Calls and returns its result. - If object has isoformat() method (e.g., datetime): Uses its ISO 8601 representation. - If object is a Path: Returns its POSIX-style path string. - If object is bytes or bytearray: Returns a hex-encoded string. - If object is an Enum: Returns its .value. - If object is a UUID: Returns its hexadecimal string. - If object is an Exception: Returns a dict with type, message, and args. - If object is a dataclass: Converts to a dictionary via asdict().

Parameters:

data (Any) -- The object to be serialized.

Returns:

A JSON-serializable object.

Return type:

Any

edri.utility.normalized_default_dict module

class edri.utility.normalized_default_dict.NormalizedDefaultDict(default_factory: Callable[[], VT] | None = None, /, *args: Any, normalization: Callable[[Any], str] | None = None, **kwargs: VT)

Bases: defaultdict[str, VT], Generic

A defaultdict variant that normalizes keys before dictionary operations.

NormalizedDefaultDict behaves like collections.defaultdict, but normalizes keys consistently before storing, retrieving, checking membership, deleting, and updating values. This is useful when keys may come from user input, external APIs, configuration files, event names, headers, or other sources where the same logical key may appear in different forms.

By default, keys must be strings and are normalized using str.lower(). A custom normalization function can be supplied via the normalization keyword argument.

Unlike a plain defaultdict, lookup operations such as d["KEY"] and d["key"] refer to the same stored value when the default normalization is used.

Parameters:

• default_factory (Callable[[], VT] | None) -- A callable used to create default values for missing keys, exactly like collections.defaultdict. If None, missing keys raise KeyError.

• args (Any) -- Optional initial data passed as a mapping or iterable of key-value pairs. All keys from the initial data are normalized before being stored.

• normalization (Callable[[Any], str] | None) -- A callable that receives the original lookup key and returns its normalized string representation. If omitted, the default normalization requires a string key and returns key.lower().

• kwargs (VT) -- Additional initial key-value pairs. Keyword names are normalized before being stored.

Raises:

• TypeError -- If normalization is not callable and not None.

• TypeError -- If the default normalization is used with a non-string key.

• KeyError -- If a key is missing and default_factory is None.

Example:

>>> d = NormalizedDefaultDict(set)
>>> d["EVENT_TYPE"].add("connection")
>>> d["event_type"]
{'connection'}
>>> "EVENT_TYPE" in d
True
>>> "event_type" in d
True

Example:

>>> d = NormalizedDefaultDict(
...     list,
...     normalization=lambda key: str(key).strip().lower(),
... )
>>> d["  User-ID  "].append(123)
>>> d["user-id"]
[123]

Note:

The normalized key is what is actually stored internally. For example, with the default normalization, assigning d["FOO"] = value stores the key as "foo".

Note:

The get() method does not create a default value for missing keys. This matches standard dict behavior and avoids triggering collections.defaultdict.__missing__().

Warning:

The normalization function should be deterministic and should preferably be idempotent. In other words, normalizing the same key multiple times should return the same result.

copy() → Self

Return a shallow copy of this normalized defaultdict.

The returned object preserves the same default_factory and normalization function. Stored keys are already normalized, so they are copied directly without re-normalization.

Returns:

Shallow copy of this instance.

get(key: Any, default: None = None, /) → VT | None

get(key: Any, default: VT, /) → VT

get(key: Any, default: _T, /) → VT | _T

Return the value for key if present, otherwise return default.

The key is normalized before lookup. Unlike __getitem__(), this method does not call __missing__() and therefore does not create a default value as a side effect.

Parameters:

• key -- Key to normalize before lookup.

• default -- Value returned when the normalized key is not present.

Returns:

Stored value or default.

pop(key: Any) → VT

pop(key: Any, default: VT) → VT

pop(key: Any, default: _T) → VT | _T

Remove and return the value associated with the normalized form of key.

If default is provided and the normalized key is missing, default is returned instead of raising KeyError.

Parameters:

• key -- Key to normalize before removal.

• default -- Optional fallback value returned when the key is missing.

Returns:

Removed value, or default if provided and the key is missing.

Raises:

KeyError -- If the normalized key is missing and no default is provided.

setdefault(key: Any) → VT | None

setdefault(key: Any, default: VT) → VT

Return the value for key, inserting default if the key is missing.

The key is normalized before lookup or insertion.

Parameters:

• key -- Key to normalize before lookup or insertion.

• default -- Value to store if the normalized key is missing.

Returns:

Existing value if present, otherwise default.

update(*args: Any, **kwargs: VT) → None

Update the dictionary with normalized keys.

Accepts the same forms as dict.update(): either one mapping, one iterable of key-value pairs, keyword arguments, or a combination of one positional source and keyword arguments.

Parameters:

• args -- Optional mapping or iterable of key-value pairs.

• kwargs -- Additional key-value pairs.

Raises:

TypeError -- If more than one positional argument is provided.

edri.utility.queue module

edri.utility.shared_memory_pipe module

class edri.utility.shared_memory_pipe.SharedMemoryPipe(name: str | None = None, *, max_slots: int = 64, slot_size: int = 262144, total_size: int | None = None, _closed: bool = False)

Bases: object

A bounded, single-producer/single-consumer byte pipe implemented on top of POSIX shared memory + POSIX semaphores.

The shared memory region is divided into max_slots fixed-size slots. Each slot contains a small header followed by up to slot_size - HEADER_SIZE bytes of payload.

Synchronization:

• slots semaphore counts free slots (writer acquires before writing).

• items semaphore counts filled slots (reader acquires before reading).

Termination:

• EOF is signaled by writing a slot whose header has eof=1.

• read() returns None on EOF; the iterator stops.

Notes

• Intended for local IPC between processes on the same host.

• The writer end owns creation of SHM and semaphores; the reader end attaches by name.

HEADER_FMT = 'I B 3x'

HEADER_SIZE = 8

close() → None

Close this endpoint.

Writer semantics:

• Enqueues an explicit EOF slot (empty payload with eof=1), then closes the shared memory handle.

Reader semantics:

• Closes the shared memory handle and unlinks the shared memory segment (removing it from the system namespace).

Notes

• This method is idempotent.

• Depending on your lifecycle expectations, you may prefer that the writer is responsible for unlinking SHM instead of the reader.

read() → bytes | None

Read the next chunk from the pipe.

Returns:

• bytes | None -- The next payload chunk, or None if an EOF slot was encountered.

• Behavior

• --------

• This method -- 1) waits for an available item (items.acquire()), 2) reads the header to determine payload length and EOF flag, 3) copies out the payload bytes, 4) advances the local ring index, 5) frees the slot (slots.release()).

Notes

Returning None is the EOF sentinel used by __next__ to stop iteration.

reader() → SharedMemoryPipe

Create a reader endpoint attached to the same underlying shared memory.

Returns:

A new instance configured as the reader end of this pipe.

Return type:

SharedMemoryPipe

Raises:

RuntimeError -- If called on an instance that is already a reader.

Notes

The returned reader shares the same SHM name and semaphore names, but maintains its own local read index.

write(data: bytes, /, *, close: bool = False) → None

Write bytes into the pipe, splitting into slot-sized chunks as needed.

Parameters:

• data -- The bytes to send.

• close -- If True, send end-of-stream after the data. All of data is written first, then a dedicated empty EOF slot is emitted (via close()), so the reader observes every byte and then stops. Reliable even when data is empty.

Notes

• Chunk size is slot_size - HEADER_SIZE.

• EOF is always its own empty slot; data slots never carry the EOF flag. (Marking a data-carrying slot as EOF would make read() drop that slot's payload, since it returns None on EOF.)

edri.utility.storage module

class edri.utility.storage.Storage(choices: LiteralString = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', length: int = 16)

Bases: dict[str, T], Generic

A generic storage class that extends a dictionary to hold items of any type, identified by unique string keys. It provides functionality to automatically generate unique keys for items if not provided.

Type Parameters:

T: The type of the items to be stored in the dictionary.

Examples:

>>> storage = Storage[int]()
>>> key1 = storage.append(100)
>>> key2 = storage.append(200, key="custom_key")
>>> storage[key1]
100
>>> storage["custom_key"]
200

Note:

• The Storage class ensures that all keys are unique. If a key is not provided, it automatically generates a unique key composed of random letters and digits.

• The default key length is 16 characters, but this can be customized during initialization.

append(item: T, key: str | None = None) → str

Adds an item to the storage with an automatically generated or specified key.

If the key already exists in the storage, a KeyError is raised to prevent overwriting existing items.

Parameters:

• item (T) -- The item to be stored.

• key (str, optional) -- The key under which to store the item. If None, a unique key is generated automatically.

Returns:

The key used to store the item in the dictionary.

Return type:

str

Raises:

KeyError -- If the specified key already exists in the storage.

Examples:

>>> storage = Storage[str]()
>>> generated_key = storage.append("apple")
>>> storage[generated_key]
'apple'
>>> custom_key = storage.append("banana", key="fruit_1")
>>> storage["fruit_1"]
'banana'
>>> storage.append("cherry", key="fruit_1")
Traceback (most recent call last):
    ...
KeyError: 'Key already exists: fruit_1'

edri.utility.transformation module

class edri.utility.transformation.StringTransformer(value: Any, transform: Callable[[str], str] | None = None, lower: bool = False, upper: bool = False)

Bases: str

A class that extends str and allows transformation on initialization. The transformation can be predefined (like lowercasing) or custom using a lambda function. The transformation settings are applied during object initialization.

Parameters:

• transform (callable, optional) -- A function that takes a string and returns a transformed version.

• lower (bool, optional) -- If True, the string will be converted to lowercase.

• upper (bool, optional) -- If True, the string will be converted to uppercase.

Example

>>> st = StringTransformer("Hello", lower=True)
>>> print(st)
hello

>>> st = StringTransformer("Hello", transform=lambda x: x[::-1])
>>> print(st)
olleH

edri.utility.validation module

class edri.utility.validation.DateTimeValidator(value: str | datetime, /, *, minimum_datetime: datetime | None = None, maximum_datetime: datetime | None = None)

Bases: datetime

A datetime type that performs validation on initialization.

This class validates the datetime against optional constraints: - Minimum allowed datetime. - Maximum allowed datetime.

Like the other validators, it accepts a single raw value (an ISO str or an existing datetime) so it can be used through inject(...).

Parameters:

• value (str | datetime) -- ISO datetime string (e.g. "2024-03-28T15:45") or a datetime.

• minimum_datetime (datetime, optional) -- The earliest valid datetime.

• maximum_datetime (datetime, optional) -- The latest valid datetime.

Raises:

ValueError -- If the value cannot be parsed or is outside the allowed bounds.

Example

>>> DateTimeValidator("2024-03-28T15:45", minimum_datetime=datetime(2024, 3, 1))
DateTimeValidator(2024, 3, 28, 15, 45)

class edri.utility.validation.DateValidator(value: str | date, /, *, minimum_date: date | None = None, maximum_date: date | None = None)

Bases: date

A date type that performs validation on initialization.

This class validates the date against optional constraints: - Minimum allowed date. - Maximum allowed date.

Like the other validators, it accepts a single raw value (an ISO str or an existing date/datetime) so it can be used through inject(...).

Parameters:

• value (str | date) -- ISO date string (e.g. "2024-03-28") or a date.

• minimum_date (date, optional) -- The earliest valid date.

• maximum_date (date, optional) -- The latest valid date.

Raises:

ValueError -- If the value cannot be parsed or is outside the allowed bounds.

Example

>>> DateValidator("2024-03-28", minimum_date=date(2024, 1, 1))
DateValidator(2024, 3, 28)

class edri.utility.validation.FloatValidator(value: Any, minimum: float | None = None, maximum: float | None = None)

Bases: float

A float type that performs validation on initialization.

This class validates the float against optional constraints: - Minimum and maximum allowed values.

Parameters:

• data (float) -- The input float to validate.

• minimum (float, optional) -- Minimum allowed value.

• maximum (float, optional) -- Maximum allowed value.

Raises:

ValueError -- If the value is less than minimum or greater than maximum.

Example

>>> FloatValidator(3.14, minimum=1.0, maximum=5.0)
3.14

class edri.utility.validation.IntegerValidator(value: Any, /, *, minimum: int | None = None, maximum: int | None = None)

Bases: int

An integer type that performs validation on initialization.

This class validates the integer against optional constraints: - Minimum and maximum allowed values.

Parameters:

• data (int) -- The input integer to validate.

• minimum (int, optional) -- Minimum allowed value.

• maximum (int, optional) -- Maximum allowed value.

Raises:

ValueError -- If the value is less than minimum or greater than maximum.

Example

>>> IntegerValidator(5, minimum=1, maximum=10)
5

class edri.utility.validation.ListValidator(iterable: Iterable[Any] = (), /, *, minimum_length: int | None = None, maximum_length: int | None = None, generics: list[Any] | None = None)

Bases: list[Any]

A list type that performs validation on initialization.

This class validates the list against optional constraints: - Minimum allowed length. - Maximum allowed length.

Parameters:

• iterable (Iterable, optional) -- Values to initialize the list with.

• minimum_length (int, optional) -- The smallest allowed list length.

• maximum_length (int, optional) -- The largest allowed list length.

Raises:

ValueError -- If the list length is outside the allowed bounds.

Example

>>> ListValidator([1, 2, 3], minimum_length=2)
[1, 2, 3]
>>> ListValidator([1, 2, 3], maximum_length=2)
Traceback (most recent call last):
    ...
ValueError: List length '3' is greater than maximum allowed '2'

class edri.utility.validation.StringValidator(value: Any, /, *, maximum_length: Any | None = None, minimum_length: int | None = None, regex: Pattern[str] | None = None)

Bases: str

A string type that performs validation on initialization.

This class validates the string against optional constraints: - A regular expression pattern. - Minimum and maximum allowed lengths. - Maximum length derived from a column-like object exposing .type.length.

Parameters:

• value (str) -- The input string to validate.

• maximum_length (int, optional) -- Maximum allowed length of the string, or a column-like object with a string type length.

• minimum_length (int, optional) -- Minimum required length of the string.

• regex (Pattern, optional) -- A compiled regex pattern the string must match.

Raises:

ValueError -- If the string does not match the regex, or its length is outside the allowed bounds.

Example

>>> StringValidator("hello", minimum_length=3, maximum_length=10)
'hello'

static safe_val(v: str, max_len: int = 20) → str

Return truncated display string, not full content.

class edri.utility.validation.TimeValidator(value: str | time, /, *, minimum_time: time | None = None, maximum_time: time | None = None)

Bases: time

A time type that performs validation on initialization.

This class validates the time against optional constraints: - Minimum allowed time. - Maximum allowed time.

Like the other validators, it accepts a single raw value (an ISO str or an existing time) so it can be used through inject(...).

Parameters:

• value (str | time) -- ISO time string (e.g. "12:30") or a time.

• minimum_time (time, optional) -- The earliest valid time.

• maximum_time (time, optional) -- The latest valid time.

Raises:

ValueError -- If the value cannot be parsed or is outside the allowed bounds.

Example

>>> TimeValidator("12:30", maximum_time=time(20, 0))
TimeValidator(12, 30)

edri.utility.watcher module

Module contents