Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/himport.py: 40%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""HIMPORT client-side fieldset registry for redis-py.
3`HIMPORT` lets a client register an ordered list of hash field names once per
4connection (a *fieldset*) and then create many hashes by sending only values.
5Because a fieldset is server-side session state bound to a single physical
6connection, redis-py keeps a client-level registry of the fieldsets the
7application has declared and prepares them lazily, per connection, on first use
8by ``himport_set``.
10This module holds only that registry. It is pure in-memory state with no I/O and
11no ``asyncio`` primitives, so a single class is shared by both the sync and async
12clients.
14Example::
16 >>> from redis.himport import HImportRegistry
17 >>> registry = HImportRegistry()
18 >>> registry.prepare("account_data", ["name", "email", "age"])
19 >>> registry.get("account_data").fields
20 ('name', 'email', 'age')
21"""
23import threading
24from collections.abc import Iterable, Iterator
25from dataclasses import dataclass
27from redis.exceptions import DataError
28from redis.typing import EncodableT, FieldT, KeyT
30# ---------------------------------------------------------------------------
31# Wire command tokens
32# ---------------------------------------------------------------------------
33# Centralised so the sync/async clients, the cluster clients and the
34# CoreCommands mixin never repeat the literal HIMPORT strings. ``HIMPORT_*`` are
35# the full command names passed to ``execute_command`` and registered as
36# response callbacks (the request packer splits the space). The
37# ``himport_*_command`` builders return the positional wire args used when a
38# connection packs commands directly (lazy PREPARE bundled with SET, deferred
39# DISCARD reconcile, etc.).
41_HIMPORT = "HIMPORT"
42_PREPARE = "PREPARE"
43_SET = "SET"
44_DISCARD = "DISCARD"
45_DISCARDALL = "DISCARDALL"
47HIMPORT_PREPARE = f"{_HIMPORT} {_PREPARE}"
48HIMPORT_SET = f"{_HIMPORT} {_SET}"
49HIMPORT_DISCARD = f"{_HIMPORT} {_DISCARD}"
50HIMPORT_DISCARDALL = f"{_HIMPORT} {_DISCARDALL}"
53def himport_prepare_command(fieldset_name: str, fields: Iterable[FieldT]) -> tuple:
54 """Positional wire args for ``HIMPORT PREPARE fieldset_name field [field ...]``."""
55 return (_HIMPORT, _PREPARE, fieldset_name, *fields)
58def himport_set_command(
59 key: KeyT, fieldset_name: str, values: Iterable[EncodableT]
60) -> tuple:
61 """Positional wire args for ``HIMPORT SET key fieldset_name value [value ...]``."""
62 return (_HIMPORT, _SET, key, fieldset_name, *values)
65def himport_discard_command(fieldset_name: str) -> tuple:
66 """Positional wire args for ``HIMPORT DISCARD fieldset_name``."""
67 return (_HIMPORT, _DISCARD, fieldset_name)
70_HIMPORT_SET_LEN = len(HIMPORT_SET)
71_HIMPORT_SET_BYTES = HIMPORT_SET.encode()
74def is_himport_set_command(command_name) -> bool:
75 """Return ``True`` if ``command_name`` names the ``HIMPORT SET`` command.
77 Redis command names are case-insensitive on the wire, and a caller using the
78 raw ``execute_command`` API may pass the name in any case and as either
79 ``str`` or ``bytes`` (e.g. ``execute_command("himport set", ...)`` or
80 ``b"HIMPORT SET"``). The connection-state-aware ``HIMPORT SET`` path keys off
81 the command name, so it must recognise all of those spellings: an exact
82 comparison against :data:`HIMPORT_SET` would miss them and send a bare SET
83 that fails with ``no such fieldset`` for a fieldset registered in the client
84 but not yet PREPAREd on the borrowed connection. The length check keeps the
85 per-command cost on the hot path to a single comparison for the common case
86 of a differently-sized command name (``upper()`` runs only on a size match).
87 """
88 if isinstance(command_name, str):
89 return (
90 len(command_name) == _HIMPORT_SET_LEN
91 and command_name.upper() == HIMPORT_SET
92 )
93 if isinstance(command_name, (bytes, bytearray, memoryview)):
94 raw = bytes(command_name)
95 return len(raw) == _HIMPORT_SET_LEN and raw.upper() == _HIMPORT_SET_BYTES
96 return False
99_HIMPORT_BYTES = _HIMPORT.encode()
100_SET_BYTES = _SET.encode()
103def _token_is(value, upper_token: str, upper_token_bytes: bytes) -> bool:
104 """Case-insensitive match of a single ``str``/``bytes`` command token.
106 ``upper_token`` / ``upper_token_bytes`` must already be upper-cased. The
107 length guard keeps the hot path to a single comparison for a differently
108 sized token (``upper()`` runs only on a size match).
109 """
110 n = len(upper_token)
111 if isinstance(value, str):
112 return len(value) == n and value.upper() == upper_token
113 if isinstance(value, (bytes, bytearray, memoryview)):
114 raw = bytes(value)
115 return len(raw) == n and raw.upper() == upper_token_bytes
116 return False
119def parse_himport_set_args(args):
120 """Detect an ``HIMPORT SET`` command in ``args`` and return its operands.
122 ``HIMPORT SET`` reaches the raw ``execute_command`` API in two wire-equivalent
123 forms that the serializer both accept:
125 * the joined form ``("HIMPORT SET", key, fieldset, *values)`` (``args[0]`` is
126 the two-word command name the request packer splits on the space), and
127 * the split form ``("HIMPORT", "SET", key, fieldset, *values)``.
129 Both are case- and encoding-insensitive. The connection-state-aware executor
130 and the pipeline pre-flight need the operands at the right offsets for either
131 form, so this returns ``(key, fieldset_name, values_list)`` when ``args`` is an
132 ``HIMPORT SET`` with enough operands, or ``None`` otherwise (a non-``HIMPORT
133 SET`` command, or one with too few operands -- which falls through to the plain
134 send so the server returns its own arity error).
135 """
136 if not args:
137 return None
138 first = args[0]
139 # Joined form: args[0] == "HIMPORT SET".
140 if is_himport_set_command(first):
141 if len(args) < 3:
142 return None
143 return args[1], args[2], list(args[3:])
144 # Split form: args[0] == "HIMPORT", args[1] == "SET".
145 if (
146 len(args) >= 2
147 and _token_is(first, _HIMPORT, _HIMPORT_BYTES)
148 and _token_is(args[1], _SET, _SET_BYTES)
149 ):
150 if len(args) < 4:
151 return None
152 return args[2], args[3], list(args[4:])
153 return None
156@dataclass(frozen=True)
157class HImportFieldset:
158 """An immutable HIMPORT fieldset entry.
160 Attributes:
161 name: Fieldset name used by ``HIMPORT SET`` / ``HIMPORT DISCARD``.
162 fields: Ordered field names, exactly as supplied by the caller. They are
163 never reordered or deduplicated (HLD R.2): the server canonicalizes
164 field order internally and rejects duplicate field names, so the
165 client only preserves the caller's positional order.
166 version: Monotonic stamp bumped each time the fieldset is (re)declared.
167 Connections compare the version they last prepared against this value
168 to detect a stale prepared state; the stamp is drawn from the registry's
169 mutation clock (:attr:`HImportRegistry.revision`), which never repeats,
170 so discarding and re-declaring the same name yields a fresh version
171 rather than a colliding one. This is the prepare-side signal; the
172 discard-side counterpart is the clock advancing on removal, since a
173 removed fieldset leaves no entry to stamp.
174 """
176 name: str
177 fields: tuple[FieldT, ...]
178 version: int
181class HImportRegistry:
182 """Client-level registry of HIMPORT fieldsets.
184 Pure in-memory state, shared by the sync and async clients. It is mutated only
185 through the client's ``himport_prepare`` / ``himport_discard`` /
186 ``himport_discard_all`` methods and exposed read-only through the client's
187 ``himport_registry`` property. Mutations are serialized under a lock so the
188 revision bump and dict change stay consistent when a sync ``Redis`` instance is
189 shared across threads. Reads that iterate or snapshot the registry take the same
190 lock, so a concurrent mutation cannot make them observe a torn view or raise
191 ``dictionary changed size during iteration``; single-key/scalar reads (``get``,
192 ``__contains__``, ``__len__``, :attr:`revision`) are atomic and stay lock-free.
193 The async client runs single-threaded and never contends.
195 The registry always starts empty; fieldsets are declared at runtime through the
196 client's ``himport_prepare`` method.
197 """
199 def __init__(self) -> None:
200 self._fieldsets: dict[str, HImportFieldset] = {}
201 # Serializes mutations (prepare/discard/discard_all) so the revision bump and
202 # the dict change are applied atomically under concurrent access from a
203 # thread-shared sync client, and guards the reads that iterate/snapshot the
204 # dict so they never see a torn view or a mid-iteration resize. Held only
205 # across synchronous, non-blocking bodies, never across I/O, so it is harmless
206 # for the single-threaded async client.
207 self._lock = threading.Lock()
208 # Monotonic mutation clock. It advances on every registry change (prepare
209 # and discard). prepare stamps the new entry with the advanced value as its
210 # per-fieldset version; discard has no surviving entry to stamp, so the
211 # advance itself is the signal that a removal happened. Because the clock
212 # only ever increases, a re-declared name never reuses a stamp, while
213 # comparing per-fieldset versions still avoids forcing unrelated fieldsets
214 # to re-prepare.
215 self._revision: int = 0
217 # -- internal helpers -------------------------------------------------
218 # ``_advance`` and ``_set`` assume ``_lock`` is already held (they run under the
219 # public mutation methods); they never acquire the lock themselves.
221 def _advance(self) -> int:
222 self._revision += 1
223 return self._revision
225 @staticmethod
226 def _materialize_fields(fields: Iterable[FieldT]) -> tuple:
227 """Validate and materialize the caller's field iterable into a tuple.
229 Done *before* the mutation lock is taken: consuming an arbitrary iterable
230 can be slow, or -- for a generator that inspects this same registry -- can
231 re-enter a locked read (e.g. ``yield`` then ``registry.names()``). Running
232 it under the non-reentrant ``_lock`` would stall every registry user or
233 deadlock permanently. Field order is preserved; nothing is reordered or
234 deduplicated.
235 """
236 # A bare single field name (str/bytes/bytearray/memoryview) is itself
237 # iterable element-by-element; that is almost certainly a caller mistake and
238 # would silently register single-character/single-byte "fields" (e.g.
239 # memoryview(b"id") -> field names 105, 100), so reject it as invalid local
240 # API usage. int/float are not iterable, so tuple() below rejects them.
241 if isinstance(fields, (str, bytes, bytearray, memoryview)):
242 raise DataError(
243 "HIMPORT fields must be a collection of field names, "
244 "not a single string or binary value"
245 )
246 field_tuple = tuple(fields)
247 if not field_tuple:
248 raise DataError("HIMPORT fieldset must have at least one field")
249 return field_tuple
251 def _set(self, name: str, field_tuple: tuple) -> HImportFieldset:
252 # ``field_tuple`` is already validated/materialized by
253 # :meth:`_materialize_fields`; only the (cheap, non-blocking) revision bump
254 # and dict mutation run here, so ``_lock`` is never held across arbitrary
255 # caller code.
256 fieldset = HImportFieldset(
257 name=name,
258 fields=field_tuple,
259 version=self._advance(),
260 )
261 self._fieldsets[name] = fieldset
262 return fieldset
264 # -- mutation ---------------------------------------------------------
266 def prepare(self, name: str, fields: Iterable[FieldT]) -> HImportFieldset:
267 """Add or replace a fieldset, bumping its version, and return the entry.
269 Re-declaring an existing name replaces its fields and bumps its version.
270 Field order is preserved verbatim; nothing is reordered or deduplicated.
271 """
272 # Materialize/validate the iterable *outside* the lock so a slow or
273 # registry-re-entrant generator can never stall or deadlock other users;
274 # the lock then covers only the atomic revision-bump + dict mutation.
275 field_tuple = self._materialize_fields(fields)
276 with self._lock:
277 return self._set(name, field_tuple)
279 def discard(self, name: str) -> bool:
280 """Remove a fieldset from the registry.
282 Returns ``True`` if a fieldset was removed, ``False`` if ``name`` was not
283 registered. Advances :attr:`revision` when a fieldset is actually removed.
284 """
285 with self._lock:
286 if name not in self._fieldsets:
287 return False
288 del self._fieldsets[name]
289 self._advance()
290 return True
292 def discard_all(self) -> int:
293 """Remove all fieldsets and return the number removed.
295 Advances :attr:`revision` when at least one fieldset is removed.
296 """
297 with self._lock:
298 count = len(self._fieldsets)
299 if count:
300 self._fieldsets.clear()
301 self._advance()
302 return count
304 # -- read-only access -------------------------------------------------
305 # Reads that iterate or build a snapshot of the dict take ``_lock`` so a
306 # concurrent mutation cannot resize it mid-iteration or expose a torn view;
307 # single-key/scalar reads below (get/__contains__/__len__/revision) are atomic
308 # and deliberately stay lock-free.
310 @property
311 def revision(self) -> int:
312 """Monotonic mutation clock, advanced on every registry change.
314 It is the discard-side counterpart to per-fieldset
315 :attr:`HImportFieldset.version`. A connection records the revision it last
316 reconciled against; when it differs, a discard (or prepare) has occurred
317 since, so the connection recomputes which of its prepared fieldsets are no
318 longer registered — see :meth:`names_to_discard` — and discards those when
319 it is released back to the pool.
320 """
321 return self._revision
323 def names_to_discard(self, prepared_names: Iterable[str]) -> list[str]:
324 """Return which of ``prepared_names`` are no longer registered.
326 Given the fieldset names a connection has prepared on the server, this is
327 the set that must be sent ``HIMPORT DISCARD`` (typically when the
328 connection is released), because they have been removed from the registry.
329 """
330 # Hold the lock across the comprehension so every membership test sees one
331 # consistent snapshot; ``prepared_names`` is an external iterable of plain
332 # strings and never calls back into the registry.
333 with self._lock:
334 return [name for name in prepared_names if name not in self._fieldsets]
336 def get(self, name: str) -> HImportFieldset | None:
337 """Return the fieldset registered under ``name``, or ``None``."""
338 return self._fieldsets.get(name)
340 def names(self) -> list[str]:
341 """Return the registered fieldset names."""
342 with self._lock:
343 return list(self._fieldsets)
345 def items(self) -> list[tuple[str, HImportFieldset]]:
346 """Return a snapshot of ``(name, fieldset)`` pairs."""
347 with self._lock:
348 return list(self._fieldsets.items())
350 def __contains__(self, name: object) -> bool:
351 return name in self._fieldsets
353 def __len__(self) -> int:
354 return len(self._fieldsets)
356 def __iter__(self) -> Iterator[str]:
357 # Snapshot under the lock and iterate the copy, so the lock is never held
358 # across caller consumption and a concurrent mutation cannot resize the
359 # underlying dict mid-iteration.
360 with self._lock:
361 return iter(list(self._fieldsets))
363 def __repr__(self) -> str:
364 with self._lock:
365 entries = list(self._fieldsets.items())
366 body = ", ".join(
367 f"{name}={list(fieldset.fields)}" for name, fieldset in entries
368 )
369 return f"{self.__class__.__name__}({body})"