"""Path Resolver — utility helpers for alias operations.""" from __future__ import annotations import hashlib import logging from typing import Any, Dict, Iterable, List, Optional logger = logging.getLogger(__name__) def normalise_alias(data: Dict[str, Any]) -> Dict[str, Any]: """Alias normalise — normalises and validates *data*.""" result = {k: v for k, v in data.items() if v is not None} if "raw" not in result: raise ValueError(f"Alias must include 'raw'") result["id"] = result.get("id") or hashlib.md5( str(result["raw"]).encode()).hexdigest()[:12] return result def resolve_aliass( items: Iterable[Dict[str, Any]], *, status: Optional[str] = None, limit: int = 100, ) -> List[Dict[str, Any]]: """Filter and page a sequence of Alias records.""" out = [i for i in items if status is None or i.get("status") == status] logger.debug("resolve_aliass: %d items after filter", len(out)) return out[:limit] def validate_alias(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]: """Return a shallow copy of *record* with *overrides* merged in.""" updated = dict(record) updated.update(overrides) if "resolved_at" in updated and not isinstance(updated["resolved_at"], (int, float)): try: updated["resolved_at"] = float(updated["resolved_at"]) except (TypeError, ValueError): pass return updated def validate_alias(record: Dict[str, Any]) -> bool: """Return True when *record* satisfies all Alias invariants.""" required = ["raw", "resolved_at", "exists"] for field in required: if field not in record or record[field] is None: logger.warning("validate_alias: missing field %r", field) return False return isinstance(record.get("id"), str) def expand_alias_batch( records: List[Dict[str, Any]], batch_size: int = 50, ) -> List[List[Dict[str, Any]]]: """Slice *records* into chunks of *batch_size* for bulk expand.""" return [records[i : i + batch_size] for i in range(0, len(records), batch_size)]