perf: remove isinstance from hash funcs, flatten task_path_str

Two micro-optimizations:

1. Remove per-element isinstance check in _xxhash_str/_uuid5_str — all
   call sites pass string parts, so encode() directly without checking.

2. Flatten task_path_str to avoid recursive calls for the common case
   of tuple elements being str or int (not nested tuples).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
John Kennedy
2026-07-27 17:01:37 +00:00
committed by John Kennedy
co-authored by Claude Opus 4.6
parent bc266572ac
commit 7f2d549edc
+20 -15
View File
@@ -551,7 +551,7 @@ PUSH_TRIGGER = (PUSH,)
class _TaskIDFn(Protocol):
def __call__(self, namespace: bytes, *parts: str | bytes) -> str:
def __call__(self, namespace: bytes, *parts: str) -> str:
pass
@@ -1426,32 +1426,37 @@ def _proc_input(
return val
def _uuid5_str(namespace: bytes, *parts: str | bytes) -> str:
def _uuid5_str(namespace: bytes, *parts: str) -> str:
"""Generate a UUID from the SHA-1 hash of a namespace and str parts."""
sha = sha1(namespace, usedforsecurity=False)
sha.update(b"".join(p.encode() if isinstance(p, str) else p for p in parts))
sha.update(b"".join(p.encode() for p in parts))
hex = sha.hexdigest()
return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
def _xxhash_str(namespace: bytes, *parts: str | bytes) -> str:
def _xxhash_str(namespace: bytes, *parts: str) -> str:
"""Generate a UUID from the XXH3 hash of a namespace and str parts."""
hex = xxh3_128_hexdigest(
namespace + b"".join(p.encode() if isinstance(p, str) else p for p in parts)
)
hex = xxh3_128_hexdigest(namespace + b"".join(p.encode() for p in parts))
return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
def task_path_str(tup: str | int | tuple) -> str:
def task_path_str(tup: str | int | tuple | list) -> str:
"""Generate a string representation of the task path."""
return (
f"~{', '.join(task_path_str(x) for x in tup)}"
if isinstance(tup, (tuple, list))
else f"{tup:010d}"
if isinstance(tup, int)
else str(tup)
)
if isinstance(tup, (tuple, list)):
parts: list[str] = []
for x in tup:
if isinstance(x, int):
parts.append(f"{x:010d}")
elif isinstance(x, (tuple, list)):
parts.append(task_path_str(x))
else:
parts.append(str(x))
return f"~{', '.join(parts)}"
elif isinstance(tup, int):
return f"{tup:010d}"
else:
return str(tup)
LAZY_ATOMIC_COUNTER_LOCK = threading.Lock()