63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
import inspect
|
|
import json
|
|
from typing import Any, Callable
|
|
|
|
|
|
def stable_repr(val: Any) -> str:
|
|
"""Stable JSON representation for cache key components."""
|
|
return json.dumps(val, sort_keys=True, default=str)
|
|
|
|
|
|
def get_canonical_name(item: Any) -> str:
|
|
"""Return canonical module.qualname for functions / callables."""
|
|
module = getattr(
|
|
item,
|
|
"__module__",
|
|
item.__class__.__module__,
|
|
)
|
|
qualname = getattr(
|
|
item,
|
|
"__qualname__",
|
|
getattr(item, "__name__", item.__class__.__name__),
|
|
)
|
|
return f"{module}.{qualname}"
|
|
|
|
|
|
def build_cache_key(
|
|
func: Callable,
|
|
prefix: str | None,
|
|
*args: Any,
|
|
**kwargs: Any,
|
|
) -> str:
|
|
"""
|
|
Build a deterministic cache key for func called with args/kwargs.
|
|
For bound methods, skips the first parameter if it's named 'self' or 'cls'.
|
|
"""
|
|
sig = inspect.signature(func)
|
|
bound = sig.bind_partial(*args, **kwargs)
|
|
bound.apply_defaults()
|
|
|
|
params = list(sig.parameters.values())
|
|
arguments = list(bound.arguments.items())
|
|
|
|
# Detect methods: if first parameter name is 'self' or 'cls' and it's provided in bound args,
|
|
# skip it when building the key.
|
|
if params:
|
|
first_name = params[0].name
|
|
if first_name in ("self", "cls") and first_name in bound.arguments:
|
|
arguments = arguments[1:]
|
|
|
|
arguments_fmt = [
|
|
f"{name}={stable_repr(val)}"
|
|
for name, val in sorted(arguments, key=lambda kv: kv[0])
|
|
]
|
|
|
|
key_parts = [
|
|
get_canonical_name(func),
|
|
*arguments_fmt,
|
|
]
|
|
|
|
if prefix is not None:
|
|
key_parts = [prefix] + key_parts
|
|
|
|
return ":".join(key_parts)
|