Utility¶

This document describes small helper classes that are used across EDRI.

`Storage`¶

Storage[T] is a small in-memory key/value store (dict[str, T]) that can generate unique random keys for stored items.

Storage use cases¶

generating short-lived lookup keys (tickets, temporary IDs)
keeping in-memory objects indexed by a random token

Key generation¶

If you do not provide a key, Storage generates one.
Keys are generated from a configurable character set (choices) and length.
The default character set is ASCII letters + digits.
The default key length is 16.
Key generation loops until it finds a key that is not currently used.

Storage API¶

append(item, key=None) -> str
- stores item
- returns the key used
- raises KeyError if you provide a key that already exists

Storage example¶

storage = Storage[int]()

k1 = storage.append(100)               # auto-generated key
k2 = storage.append(200, key="job")   # explicit key

assert storage[k1] == 100
assert storage["job"] == 200

Storage gotchas¶

Storage is an in-memory dict. It is not persistent.
If you need cryptographic-strength tokens, consider using a cryptographic RNG (this implementation uses random.choice).

`Queue`¶

Queue[T] extends Python’s queue.Queue by adding unget(...), which inserts an item at the front of the queue.

Queue use cases¶

re-queueing an item that should be processed again next
implementing simple reprioritization within a FIFO queue

Queue API¶

unget(item, block=True, timeout=None) -> None
- behaves like put(...) regarding blocking/full-queue behavior
- but inserts at the front instead of the end

Queue example¶

q = Queue[int]()
q.put(1)
q.put(2)

q.unget(0)     # will be processed next

assert q.get() == 0
assert q.get() == 1
assert q.get() == 2

Queue gotchas¶

If the queue has a maxsize and is full, unget can block or raise queue.Full, consistent with put.
unget increments unfinished_tasks, just like put.

`NormalizedDefaultDict`¶

NormalizedDefaultDict[VT] is a defaultdict variant that normalizes keys on both set and get. Keys are always normalized to str, so the class takes a single type parameter for the value type (VT).

NormalizedDefaultDict use cases¶

storing values indexed by case-insensitive keys (e.g. event names, resource names)
preventing accidental duplicates such as "EVENT" vs "event"

Normalization¶

Default normalization is key.lower().
You can pass a custom normalization=... function.
If you initialize from a dict, the initial keys are normalized too.

NormalizedDefaultDict example¶

d = NormalizedDefaultDict(set)   # default_factory is positional-only

d["EVENT_TYPE"].add("client-1")
d["event_type"].add("client-2")

assert d["Event_Type"] == {"client-1", "client-2"}

Custom normalization example¶

d = NormalizedDefaultDict(
    list,   # default_factory is positional-only
    normalization=lambda x: str(x).strip().lower(),
)

d["  Hello "].append(1)
assert d["hello"] == [1]

NormalizedDefaultDict gotchas¶

The default normalization assumes the key supports .lower().
- In practice, this is intended for string keys.
- If you want to accept non-string keys, provide a normalization function that converts to string first.