msyavuz commented on code in PR #44892:
URL: https://github.com/apache/superset/pull/44892#discussion_r4195697722
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +914,396 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...]:
+ """Materialize the provider's payload, or reject its shape.
+
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+
+ An exception raised *while reading* a lazy payload is left to propagate.
+ Walking a generator runs host code just as calling the provider does, so
+ the two are the same kind of failure and earn the same retry; only the
+ shape of a payload that arrived intact is settled for good.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return ()
+ return tuple(provided)
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ # Version shadows are resolved by bare name as well, in
+ # _entity_version_targets.
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ or dependency.classification is DependencyClassification.VERSION_OWNED
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Two different reasons, both about identities this package owns.
+ # A stock listener action decides what to delete from a built-in
+ # entity type, so declared by a host it is accepted and then never
+ # runs. A blocker *would* be applied -- validate_deletion_allowed
+ # is generic over the declarations -- but its reason code lands in
+ # the purge audit record, whose vocabulary is a closed set pinned
+ # by test. A host refuses a purge from its own validator instead,
+ # where the code it records is plainly its own.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies are not available
"
+ "to a host root"
+ )
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...] | None:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+
+ Returns ``None`` for the failures that may be transient -- the provider
+ could not be called, or a lazy payload raised while being read. The caller
+ then publishes the built-in roots with a short deadline, so the next
+ resolution after it tries again rather than treating one bad moment as the
+ answer for the life of the process. A payload that arrived intact but is
+ the wrong shape is a host bug that will not fix itself, so it is published
+ as an empty result instead.
+ """
+ if provider is None:
+ return ()
+ try:
+ # Both the call and reading what it returns run host code, so a lazy
+ # payload that raises mid-walk is the same kind of failure as a
+ # provider that raises outright, and earns the same retry.
+ candidates: tuple[Any, ...] = _host_policy_payload(provider())
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
+
+
+#: Sentinel distinguishing "no index resolved yet" from an index resolved for
+#: no provider, which is the ordinary case.
+_UNRESOLVED: Any = object()
+
+
+@dataclass
+class _ResolvedRegistry:
+ """The index built for one provider, with the roots validated so far.
+
+ Matched to its provider by *identity* rather than keyed in an
+ ``lru_cache``: a cache key would have to be hashed, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That raises
+ before the host boundary can isolate the host's mistake, which would
+ stop the built-in roots purging too.
+ """
+
+ provider: Any = _UNRESOLVED
+ registry: Mapping[type[Any], PurgeEntityPolicy] = field(
+ default_factory=lambda: MappingProxyType({})
+ )
+ validated: set[type[Any]] = field(default_factory=set)
+ #: Set only on a snapshot published because the provider could not be
+ #: called. Until this deadline the snapshot answers reads; after it, the
+ #: provider is tried again.
+ retry_after: float | None = None
+
+
+#: Key the resolved index lives under on the Flask app. Per app rather than
+#: per process: two apps in one process have different configs, and a single
+#: global snapshot would thrash between them -- re-resolving, and re-invoking
+#: the provider, on every read.
+_REGISTRY_EXTENSION_KEY: str = "deletion_retention_purge_registry"
+
+#: How long a snapshot published after a provider failure answers reads
+#: before the provider is tried again. Bounds both extremes: a failure is
+#: neither remembered for the life of the process nor retried on every read --
+#: at roughly three reads per purged entity, retrying per read meant tens of
+#: thousands of provider calls and tracebacks in a single pass.
+_PROVIDER_RETRY_SECONDS: float = 60.0
+
+#: Per-thread marker for "this thread is already resolving". A host provider
+#: may build its policy from a built-in one -- ``replace(get_purge_policy(
+#: Slice), model=HostRoot, ...)`` is the obvious way to borrow the stock
+#: callbacks -- which re-enters resolution while the first call is still in
+#: the provider. The nested call is handed the state already published
+#: instead, which always carries the built-in roots.
+_RESOLVING: threading.local = threading.local()
+
+
+def _resolved_registry() -> _ResolvedRegistry:
+ """Return the state resolved for the installed provider, building it once.
+
+ Readers take a snapshot and a rebuild publishes a new one, so a caller
+ always sees an index and its validation record together. Publishing is a
+ single rebind, which is what makes it safe to do this without a lock: two
+ concurrent first uses may each invoke the provider and each publish, but
+ the two snapshots are equivalent and neither is ever seen half-built.
+ Paying for a duplicate provider call is the deliberate trade -- a lock
+ here deadlocks the moment a provider resolves a built-in policy of its
+ own, and a hung purge is worse than a repeated call.
+ """
+ if not has_app_context():
+ return _builtin_registry()
+ provider: Callable[[], Any] | None = current_app.config.get(
+ HOST_POLICIES_CONFIG_KEY
+ )
+ published: _ResolvedRegistry = current_app.extensions.setdefault(
+ _REGISTRY_EXTENSION_KEY, _builtin_registry()
+ )
+ if published.provider is provider and (
+ published.retry_after is None or time.monotonic() <
published.retry_after
+ ):
+ return published
+ if getattr(_RESOLVING, "active", False):
Review Comment:
The re-entrancy guard returns the built-in registry whenever a thread is
resolving. A host policy that calls `get_purge_policy` for its own model inside
the provider gets `Unsupported purge model`. Also, `_RESOLVING.active` is reset
to False rather than restored, so it is only safe at one nesting level.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +914,396 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...]:
+ """Materialize the provider's payload, or reject its shape.
+
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+
+ An exception raised *while reading* a lazy payload is left to propagate.
+ Walking a generator runs host code just as calling the provider does, so
+ the two are the same kind of failure and earn the same retry; only the
+ shape of a payload that arrived intact is settled for good.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return ()
+ return tuple(provided)
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ # Version shadows are resolved by bare name as well, in
+ # _entity_version_targets.
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ or dependency.classification is DependencyClassification.VERSION_OWNED
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Two different reasons, both about identities this package owns.
+ # A stock listener action decides what to delete from a built-in
+ # entity type, so declared by a host it is accepted and then never
+ # runs. A blocker *would* be applied -- validate_deletion_allowed
+ # is generic over the declarations -- but its reason code lands in
+ # the purge audit record, whose vocabulary is a closed set pinned
+ # by test. A host refuses a purge from its own validator instead,
+ # where the code it records is plainly its own.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies are not available
"
+ "to a host root"
+ )
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...] | None:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+
+ Returns ``None`` for the failures that may be transient -- the provider
+ could not be called, or a lazy payload raised while being read. The caller
+ then publishes the built-in roots with a short deadline, so the next
+ resolution after it tries again rather than treating one bad moment as the
+ answer for the life of the process. A payload that arrived intact but is
+ the wrong shape is a host bug that will not fix itself, so it is published
+ as an empty result instead.
+ """
+ if provider is None:
+ return ()
+ try:
+ # Both the call and reading what it returns run host code, so a lazy
+ # payload that raises mid-walk is the same kind of failure as a
+ # provider that raises outright, and earns the same retry.
+ candidates: tuple[Any, ...] = _host_policy_payload(provider())
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
+
+
+#: Sentinel distinguishing "no index resolved yet" from an index resolved for
+#: no provider, which is the ordinary case.
+_UNRESOLVED: Any = object()
+
+
+@dataclass
+class _ResolvedRegistry:
+ """The index built for one provider, with the roots validated so far.
+
+ Matched to its provider by *identity* rather than keyed in an
+ ``lru_cache``: a cache key would have to be hashed, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That raises
+ before the host boundary can isolate the host's mistake, which would
+ stop the built-in roots purging too.
+ """
+
+ provider: Any = _UNRESOLVED
+ registry: Mapping[type[Any], PurgeEntityPolicy] = field(
+ default_factory=lambda: MappingProxyType({})
+ )
+ validated: set[type[Any]] = field(default_factory=set)
+ #: Set only on a snapshot published because the provider could not be
+ #: called. Until this deadline the snapshot answers reads; after it, the
+ #: provider is tried again.
+ retry_after: float | None = None
+
+
+#: Key the resolved index lives under on the Flask app. Per app rather than
+#: per process: two apps in one process have different configs, and a single
+#: global snapshot would thrash between them -- re-resolving, and re-invoking
+#: the provider, on every read.
+_REGISTRY_EXTENSION_KEY: str = "deletion_retention_purge_registry"
+
+#: How long a snapshot published after a provider failure answers reads
+#: before the provider is tried again. Bounds both extremes: a failure is
+#: neither remembered for the life of the process nor retried on every read --
+#: at roughly three reads per purged entity, retrying per read meant tens of
+#: thousands of provider calls and tracebacks in a single pass.
+_PROVIDER_RETRY_SECONDS: float = 60.0
+
+#: Per-thread marker for "this thread is already resolving". A host provider
+#: may build its policy from a built-in one -- ``replace(get_purge_policy(
+#: Slice), model=HostRoot, ...)`` is the obvious way to borrow the stock
+#: callbacks -- which re-enters resolution while the first call is still in
+#: the provider. The nested call is handed the state already published
+#: instead, which always carries the built-in roots.
+_RESOLVING: threading.local = threading.local()
+
+
+def _resolved_registry() -> _ResolvedRegistry:
+ """Return the state resolved for the installed provider, building it once.
+
+ Readers take a snapshot and a rebuild publishes a new one, so a caller
+ always sees an index and its validation record together. Publishing is a
+ single rebind, which is what makes it safe to do this without a lock: two
+ concurrent first uses may each invoke the provider and each publish, but
+ the two snapshots are equivalent and neither is ever seen half-built.
+ Paying for a duplicate provider call is the deliberate trade -- a lock
+ here deadlocks the moment a provider resolves a built-in policy of its
+ own, and a hung purge is worse than a repeated call.
+ """
+ if not has_app_context():
+ return _builtin_registry()
+ provider: Callable[[], Any] | None = current_app.config.get(
+ HOST_POLICIES_CONFIG_KEY
+ )
+ published: _ResolvedRegistry = current_app.extensions.setdefault(
+ _REGISTRY_EXTENSION_KEY, _builtin_registry()
+ )
+ if published.provider is provider and (
Review Comment:
Registry state is read and published on the shared app extensions without a
lock, and `validated` is a mutable set shared across threads. Concurrent first
use can run the provider twice, and `current_app.extensions.setdefault(...,
_builtin_registry())` evaluates eagerly on every call.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1341,285 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ The shared cleanup empties associations before owned children, while an
+ owned table's predicate selects its rows *through* its ownership path. If
+ a hop on that path is an association, its rows are already gone when the
+ owned delete runs, so the statement matches nothing.
+
+ Kept where the rest of the ordering rules were dropped because this one
+ can be silent: where foreign keys are enforced the association delete
+ fails loudly, but where they are not -- SQLite -- the root is purged and
+ its descendants are left orphaned with nothing reported. The remedy is to
+ classify the intermediate table as owned, which places it in the same
+ phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
- raise RuntimeError(
- f"Missing listener action for {dependency.key.describe()}"
- )
- if (
- dependency.classification is DependencyClassification.VERSION_OWNED
- and dependency.version_column is None
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage, and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _validate_recursive_ownership(policy: PurgeEntityPolicy) -> None:
+ """Reject a self-referencing owned table under the stock cleanup.
+
+ ``delete_owned_children`` issues one statement per declared edge, so a
+ table that owns itself is pruned one level deep. Where foreign keys are
+ enforced the root's own delete then fails and rolls back; where they are
+ not -- SQLite -- the root is purged and its grandchildren are left behind
+ with a dangling parent id and nothing reported. A host declaring a tree
+ supplies cleanup that walks it.
+ """
+ for dependency in policy.dependencies:
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ key: DependencyKey = dependency.key
+ if key.owner_table != key.related_table:
+ continue
+ raise RuntimeError(
+ f"Owned dependency {key.describe()} is self-referencing; the stock
"
+ "owned-child cleanup deletes one level, so this policy must supply
"
+ "its own delete_owned_children"
+ )
+
+
+def _validate_scanner_requirements(policy: PurgeEntityPolicy) -> None:
+ """Reject a root the scheduled scan cannot page through.
+
+ The retention task selects, windows and orders eligible rows by ``id``,
+ and the cascade pins each row by it. A root keyed on something else -- a
+ UUID primary key with no ``id`` column -- raises inside the scan, outside
+ the per-row error handling, so the run aborts before the remaining roots
+ are reached.
+ """
+ mapper: Mapper[Any] = sa.inspect(policy.model)
+ table: sa.Table = mapper.local_table
+ if "id" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'id' column; the "
+ "scheduled scan pages eligible rows by id"
+ )
+ primary_key: tuple[sa.Column[Any], ...] = tuple(mapper.primary_key)
+ if (
+ len(primary_key) != 1
+ or primary_key[0].name != "id"
+ or not isinstance(primary_key[0].type, sa.Integer)
+ ):
+ # The scan pages by id and the row is then fetched with a scalar
+ # Session.get, so a composite or differently named key matches
+ # nothing: every row fails while a dry run still counts it eligible.
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} is keyed on "
+ f"({', '.join(column.name for column in primary_key)}); the "
+ "scheduled purge pages an integer 'id' watermark and looks rows "
+ "up by it"
+ )
+ if "deleted_at" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'deleted_at' column; "
+ "every purge path requires the row to be archived first"
+ )
+
+
+#: Classifications the shared cleanup executes as SQL, through
+#: ``_dependency_predicates`` -- which only knows how to read an inbound
+#: foreign key.
+_EXECUTABLE_CLASSIFICATIONS: frozenset[DependencyClassification] = frozenset(
+ {
+ DependencyClassification.OWNED,
+ DependencyClassification.ASSOCIATION,
+ DependencyClassification.BLOCK,
+ }
+)
+
+
+def _validate_versioned_root(policy: PurgeEntityPolicy) -> None:
+ """Reject version targets on a root that has no version class.
+
+ ``_delete_version_history`` resolves the *root's* version class first and
+ returns as soon as that raises, so a target declared on a child's shadow
+ is never reached for an unversioned root: the root and its live child rows
+ are purged while the child's history survives, with nothing reported.
+
+ Skipped where nothing is versioned at all. The version cascade is then
+ inert for every root, including this package's own, and refusing would
+ only take those down with it.
+ """
+ try:
+ from sqlalchemy_continuum import version_class, versioning_manager
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+ except ImportError: # pragma: no cover - versioning not installed
+ return
+ if not getattr(versioning_manager, "version_class_map", None):
Review Comment:
When `version_class_map` is empty, the unversioned-root check is skipped.
Versioning set up lazily after admission would then leave the check unenforced
for good, because validated roots are never re-checked.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +914,396 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...]:
+ """Materialize the provider's payload, or reject its shape.
+
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+
+ An exception raised *while reading* a lazy payload is left to propagate.
+ Walking a generator runs host code just as calling the provider does, so
+ the two are the same kind of failure and earn the same retry; only the
+ shape of a payload that arrived intact is settled for good.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return ()
+ return tuple(provided)
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ # Version shadows are resolved by bare name as well, in
+ # _entity_version_targets.
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ or dependency.classification is DependencyClassification.VERSION_OWNED
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
Review Comment:
The built-in root check compares by model class only. A host policy with a
different class mapped onto a built-in table (e.g. a subclass of Slice) passes
admission and gets a host cleanup path against core tables.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1341,285 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ The shared cleanup empties associations before owned children, while an
+ owned table's predicate selects its rows *through* its ownership path. If
+ a hop on that path is an association, its rows are already gone when the
+ owned delete runs, so the statement matches nothing.
+
+ Kept where the rest of the ordering rules were dropped because this one
+ can be silent: where foreign keys are enforced the association delete
+ fails loudly, but where they are not -- SQLite -- the root is purged and
+ its descendants are left orphaned with nothing reported. The remedy is to
+ classify the intermediate table as owned, which places it in the same
+ phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
- raise RuntimeError(
- f"Missing listener action for {dependency.key.describe()}"
- )
- if (
- dependency.classification is DependencyClassification.VERSION_OWNED
- and dependency.version_column is None
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage, and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _validate_recursive_ownership(policy: PurgeEntityPolicy) -> None:
+ """Reject a self-referencing owned table under the stock cleanup.
+
+ ``delete_owned_children`` issues one statement per declared edge, so a
+ table that owns itself is pruned one level deep. Where foreign keys are
+ enforced the root's own delete then fails and rolls back; where they are
+ not -- SQLite -- the root is purged and its grandchildren are left behind
+ with a dangling parent id and nothing reported. A host declaring a tree
+ supplies cleanup that walks it.
+ """
+ for dependency in policy.dependencies:
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ key: DependencyKey = dependency.key
+ if key.owner_table != key.related_table:
+ continue
+ raise RuntimeError(
+ f"Owned dependency {key.describe()} is self-referencing; the stock
"
+ "owned-child cleanup deletes one level, so this policy must supply
"
+ "its own delete_owned_children"
+ )
+
+
+def _validate_scanner_requirements(policy: PurgeEntityPolicy) -> None:
+ """Reject a root the scheduled scan cannot page through.
+
+ The retention task selects, windows and orders eligible rows by ``id``,
+ and the cascade pins each row by it. A root keyed on something else -- a
+ UUID primary key with no ``id`` column -- raises inside the scan, outside
+ the per-row error handling, so the run aborts before the remaining roots
+ are reached.
+ """
+ mapper: Mapper[Any] = sa.inspect(policy.model)
+ table: sa.Table = mapper.local_table
+ if "id" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'id' column; the "
+ "scheduled scan pages eligible rows by id"
+ )
+ primary_key: tuple[sa.Column[Any], ...] = tuple(mapper.primary_key)
+ if (
+ len(primary_key) != 1
+ or primary_key[0].name != "id"
+ or not isinstance(primary_key[0].type, sa.Integer)
+ ):
+ # The scan pages by id and the row is then fetched with a scalar
+ # Session.get, so a composite or differently named key matches
+ # nothing: every row fails while a dry run still counts it eligible.
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} is keyed on "
+ f"({', '.join(column.name for column in primary_key)}); the "
+ "scheduled purge pages an integer 'id' watermark and looks rows "
+ "up by it"
+ )
+ if "deleted_at" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'deleted_at' column; "
+ "every purge path requires the row to be archived first"
+ )
+
+
+#: Classifications the shared cleanup executes as SQL, through
+#: ``_dependency_predicates`` -- which only knows how to read an inbound
+#: foreign key.
+_EXECUTABLE_CLASSIFICATIONS: frozenset[DependencyClassification] = frozenset(
+ {
+ DependencyClassification.OWNED,
+ DependencyClassification.ASSOCIATION,
+ DependencyClassification.BLOCK,
+ }
+)
+
+
+def _validate_versioned_root(policy: PurgeEntityPolicy) -> None:
+ """Reject version targets on a root that has no version class.
+
+ ``_delete_version_history`` resolves the *root's* version class first and
+ returns as soon as that raises, so a target declared on a child's shadow
+ is never reached for an unversioned root: the root and its live child rows
+ are purged while the child's history survives, with nothing reported.
+
+ Skipped where nothing is versioned at all. The version cascade is then
+ inert for every root, including this package's own, and refusing would
+ only take those down with it.
+ """
+ try:
+ from sqlalchemy_continuum import version_class, versioning_manager
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+ except ImportError: # pragma: no cover - versioning not installed
+ return
+ if not getattr(versioning_manager, "version_class_map", None):
+ return
+ try:
+ version_class(policy.model)
+ except ClassNotVersioned:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no version class, so the "
+ "version cascade returns before reaching a declared target; such "
+ "a policy declares none"
+ ) from None
+
+
+def _validate_version_target(
+ policy: PurgeEntityPolicy, dependency: DependencyPolicy, metadata:
sa.MetaData
+) -> None:
+ """Reject a version target whose column does not hold the root's id.
+
+ Version cleanup compares the declared column with the root's id directly,
+ so the column has to be root-relative. Shadow tables are generated and may
+ not exist when a policy is validated, so the check reads their live
+ counterpart -- ``<table>_version`` describes ``<table>``, the naming the
+ shadow declarations already rely on -- and accepts the column when it is
+ the root's own primary key or a foreign key into the root.
+
+ A column naming an intermediate owner, a stage id under a workflow root
+ say, matches history belonging to a different root and leaves this one's
+ behind.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ column: str = cast(str, dependency.version_column)
+ shadow_name: str = dependency.key.related_table
+ suffix: str = "_version"
+ live_name: str = (
+ shadow_name[: -len(suffix)] if shadow_name.endswith(suffix) else
shadow_name
+ )
+ live_table: sa.Table | None = metadata.tables.get(live_name)
+ if live_table is not None and column in live_table.c:
+ live_column: sa.Column[Any] = live_table.c[column]
+ primary_key: tuple[sa.Column[Any], ...] =
tuple(root_table.primary_key.columns)
+ if live_table is root_table:
+ # On the root's own shadow only its key identifies it. A
+ # self-referencing column such as ``parent_id`` points at *other*
+ # rows of the same root, so cleanup would delete the children's
+ # history and leave the purged row's own behind.
+ if len(primary_key) == 1 and live_column.name ==
primary_key[0].name:
+ return
+ # Elsewhere the comparison is still against the root's id, so a
+ # foreign key to any other root column matches rows belonging to a
+ # different root.
+ elif len(primary_key) == 1 and any(
+ foreign_key.column.table is root_table
+ and foreign_key.column.name == primary_key[0].name
+ for foreign_key in live_column.foreign_keys
):
- raise RuntimeError(
- f"Missing version target column for
{dependency.key.describe()}"
- )
+ return
+ raise RuntimeError(
+ f"Version target {shadow_name}.{column} for
{dependency.key.describe()} "
+ f"is not root-relative; cleanup compares it with the "
+ f"{root_table.name} id, so it must be that id or a foreign key to it"
+ )
+
+
+def _validate_dependency_declaration(
+ policy: PurgeEntityPolicy,
+ dependency: DependencyPolicy,
+ metadata: sa.MetaData,
+) -> None:
+ """Reject one declaration whose mistake would not announce itself."""
+ key: DependencyKey = dependency.key
+ if (
+ dependency.classification is DependencyClassification.LISTENER_EFFECT
+ and dependency.listener_action is None
+ ):
+ raise RuntimeError(f"Missing listener action for {key.describe()}")
+ if dependency.classification is DependencyClassification.VERSION_OWNED:
+ if dependency.version_column is None:
+ raise RuntimeError(f"Missing version target column for
{key.describe()}")
+ # Both silent either way: a target the version cascade never
+ # reaches, and one that is not root-relative and so deletes another
+ # root's history while leaving this root's behind.
+ _validate_versioned_root(policy)
+ _validate_version_target(policy, dependency, metadata)
+
+
+def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
+ """Reject a declaration that would delete the wrong rows, and little else.
+
+ Validation here is deliberately not an attempt to prove that an arbitrary
+ declaration executes correctly -- the space of shapes a host might write
+ is not enumerable, and every rule written to anticipate one is a rule to
+ maintain. The line drawn instead:
+
+ * a declaration whose mistake would **delete the wrong rows, or silently
+ skip cleanup**, is refused, because neither announces itself;
+ * a declaration that merely **fails loudly** is left to fail. A predicate
+ the cleanup cannot build raises per row and is counted as a cascade
+ failure; an ordering a foreign key refuses surfaces as an integrity
+ error, which the cascade reports as a blocked purge carrying the
+ ``cascade_integrity_failure`` reason code. Either way the attempt is
+ recorded, isolated to its own root by the scheduled task, and the
+ entity is retried on the next run.
+
+ The frame's own requirements are checked regardless, since the scan and
+ the locked claim read them before any policy code runs.
+ """
+ metadata: sa.MetaData = sa.inspect(policy.model).local_table.metadata
+ for dependency in policy.dependencies:
+ _validate_dependency_declaration(policy, dependency, metadata)
+ _validate_scanner_requirements(policy)
+ if policy.delete_owned_children is delete_owned_children:
+ # The hazard is the stock owned cleanup's: it builds each predicate by
+ # traversing the ownership path, which is empty by then whoever
+ # deleted the association rows. A policy that walks its own tree is
+ # not exposed to it; one that merely replaces the association delete
+ # still is.
+ _validate_owned_traversal(policy)
+ _validate_recursive_ownership(policy)
def get_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
"""Resolve a complete policy or reject an unsupported purge model."""
- return _validated_purge_policy(cast(Hashable, model))
+ resolved: _ResolvedRegistry = _resolved_registry()
+ try:
+ policy: PurgeEntityPolicy = resolved.registry[model]
+ except KeyError as ex:
+ raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ if model not in resolved.validated:
+ _validated_policy(policy)
Review Comment:
`get_purge_policy` validates built-ins lazily and mutates
`resolved.validated` after the check. When the validation raises, it is re-run
on every call, which could be once per row, and the traceback repeats each time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
