pyreorder ¶

def load(self) -> None:
    """Read the cache file. Missing or corrupt files are treated as empty.

    Old-format caches (plain hash strings) are migrated to the new
    ``{hash, seen}`` format with ``seen`` set to *now*.  Entries whose
    ``seen`` timestamp is older than ``ttl_days`` days are dropped.
    """
    if self._path is None or not self._path.exists():
        self._entries = {}
        return
    try:
        data = json.loads(self._path.read_text(encoding="utf-8"))
        if not isinstance(data, dict):
            self._entries = {}
            return
        now = time.time()
        migrated: dict[str, CacheEntry] = {}
        for k, v in data.items():
            if not isinstance(k, str):
                continue
            if isinstance(v, str):
                # Old format: plain hash string -> migrate.
                migrated[k] = {"hash": v, "seen": now}
            elif isinstance(v, dict) and "hash" in v and "seen" in v:
                migrated[k] = v
        self._entries = migrated
    except (json.JSONDecodeError, OSError):
        # Corrupt cache: silently start fresh. Better a few redundant sorts
        # than a crashed run.
        self._entries = {}
        return
    # Prune expired entries.
    if self._ttl_days > 0:
        cutoff = time.time() - self._ttl_days * 86400
        before = len(self._entries)
        self._entries = {k: v for k, v in self._entries.items() if v.get("seen", 0) >= cutoff}
        if len(self._entries) != before:
            self._dirty = True

def lookup(self, config_sig: str, source_hash: str, current_content_hash: str) -> bool:
    """True when the file can be skipped (already in sorted state).

    A hit requires the stored sorted-output hash to match the file's current
    content hash -- otherwise the file was edited after caching and must sort.
    On a hit the ``seen`` timestamp is refreshed.
    """
    key = f"{config_sig}:{source_hash}"
    entry = self._entries.get(key)
    if entry is None or entry.get("hash") != current_content_hash:
        return False
    # Refresh last-seen timestamp.
    entry["seen"] = time.time()
    self._dirty = True
    return True

def merge(self, entries: dict[str, CacheEntry]) -> None:
    """Merge external cache entries (e.g. from parallel workers) into this cache."""
    for key, value in entries.items():
        if self._entries.get(key) != value:
            self._entries[key] = value
            self._dirty = True

def record(self, config_sig: str, source_hash: str, sorted_hash: str) -> None:
    """Record (or refresh) the sorted-output hash for a source hash."""
    key = f"{config_sig}:{source_hash}"
    existing = self._entries.get(key)
    if existing is None or existing.get("hash") != sorted_hash:
        self._entries[key] = {"hash": sorted_hash, "seen": time.time()}
        self._dirty = True

def save(self) -> None:
    """Atomically write the cache if it changed. No-op without a path."""
    if self._path is None or not self._dirty:
        return
    self._path.parent.mkdir(parents=True, exist_ok=True)
    tmp = self._path.with_suffix(self._path.suffix + ".tmp")
    payload: dict[str, CacheEntry] = self._entries
    tmp.write_text(json.dumps(payload), encoding="utf-8")
    tmp.replace(self._path)
    self._dirty = False

def config_signature(self) -> str:
    """Stable hash of the output-affecting fields + pyreorder version.

    Used as part of the content-hash cache key. Changes to any field that
    affects sorted output, or to the pyreorder version, invalidate the cache.
    Non-output fields (``fail_on_changed``, ``exclude``, ``recursive``,
    ``unknown_section``, ``cache_*``, ``config_path``) are excluded.
    """
    try:
        ver = _pkg_version("pyreorder")
    except PackageNotFoundError:
        ver = "0.0.0"
    relevant = {
        "sections": self.sections,
        "strategies": self.strategies,
        "class_methods_enabled": self.class_methods_enabled,
        "class_methods_order": self.class_methods_order,
        "class_methods_type_order": self.class_methods_type_order,
        "constants_pattern": self.constants_pattern,
        "dunder_exports_names": self.dunder_exports_names,
        "hoist_inline_imports": self.hoist_inline_imports,
        "hoist_main_imports": self.hoist_main_imports,
        "remove_type_checking": self.remove_type_checking,
        "version": ver,
    }
    payload = _json_dumps(relevant, sort_keys=True).encode()
    return sha256(payload).hexdigest()[:16]

@classmethod
def from_table(  # noqa: PLR0915
    cls,
    data: dict[str, Any],
    *,
    path: Path | None = None,
    raw: dict[str, Any] | None = None,
    is_pyproject: bool = False,
) -> Config:
    """Build a :class:`Config` from a parsed ``pyreorder`` table."""
    cfg = cls(config_path=path)

    module = data.get("module", {}) or {}
    if isinstance(module, dict) and "sections" in module:
        sections = module["sections"]
        if isinstance(sections, list) and all(isinstance(s, str) for s in sections):
            cfg.sections = [s for s in sections if isinstance(s, str)]

    strategy = data.get("strategy", {}) or {}
    if isinstance(strategy, dict):
        for name, value in strategy.items():
            if not isinstance(name, str):
                continue
            if value in VALID_STRATEGIES:
                cfg.strategies[name] = value
            else:
                warnings.warn(
                    f"pyreorder: unknown strategy {value!r} for section {name!r}; ignoring",
                    stacklevel=2,
                )

    cm = data.get("class_methods", {}) or {}
    if isinstance(cm, dict) and cm:
        cfg._apply_class_methods(cm)
    elif raw is not None:
        cfg._apply_class_methods(_legacy_undersort_overrides(raw, is_pyproject=is_pyproject))

    classification = data.get("classification", {}) or {}
    if isinstance(classification, dict):
        if "constants_pattern" in classification:
            pattern = classification["constants_pattern"]
            if isinstance(pattern, str):
                cfg.constants_pattern = pattern
        if "dunder_exports_names" in classification:
            names = classification["dunder_exports_names"]
            if isinstance(names, list) and all(isinstance(n, str) for n in names):
                cfg.dunder_exports_names = list(names)
            else:
                warnings.warn(
                    "pyreorder: classification.dunder_exports_names must be a list of strings; ignoring",
                    stacklevel=2,
                )

    transforms = data.get("transforms", {}) or {}
    if isinstance(transforms, dict):
        if "hoist_inline_imports" in transforms:
            cfg.hoist_inline_imports = bool(transforms["hoist_inline_imports"])
        if "hoist_main_imports" in transforms:
            cfg.hoist_main_imports = bool(transforms["hoist_main_imports"])
        if "remove_type_checking" in transforms:
            cfg.remove_type_checking = bool(transforms["remove_type_checking"])

    cli = data.get("cli", {}) or {}
    if isinstance(cli, dict):
        if "fail_on_changed" in cli:
            cfg.fail_on_changed = bool(cli["fail_on_changed"])
        if "cache" in cli:
            cfg.cache_enabled = bool(cli["cache"])
        if "cache_dir" in cli:
            cache_dir = cli["cache_dir"]
            if isinstance(cache_dir, str):
                cfg.cache_dir = Path(cache_dir)
            else:
                warnings.warn(
                    "pyreorder: cli.cache_dir must be a string; ignoring",
                    stacklevel=2,
                )
        if "jobs" in cli:
            jobs = cli["jobs"]
            if isinstance(jobs, int):
                cfg.jobs = jobs
            else:
                warnings.warn(
                    "pyreorder: cli.jobs must be an integer; ignoring",
                    stacklevel=2,
                )
        if "parallel_backend" in cli:
            backend = cli["parallel_backend"]
            if isinstance(backend, str) and backend in ("process", "thread"):
                cfg.parallel_backend = backend
            else:
                warnings.warn(
                    "pyreorder: cli.parallel_backend must be 'process' or 'thread'; ignoring",
                    stacklevel=2,
                )

    if "cache_ttl_days" in cli:
        ttl = cli["cache_ttl_days"]
        if isinstance(ttl, int) and ttl >= 0:
            cfg.cache_ttl_days = ttl
        else:
            warnings.warn(
                "pyreorder: cli.cache_ttl_days must be a non-negative int; ignoring",
                stacklevel=2,
            )

    discovery = data.get("discovery", {}) or {}
    if isinstance(discovery, dict):
        if "exclude" in discovery:
            exclude = discovery["exclude"]
            if isinstance(exclude, list) and all(isinstance(p, str) for p in exclude):
                cfg.exclude = [str(p) for p in exclude]
            else:
                warnings.warn(
                    "pyreorder: discovery.exclude must be a list of strings; ignoring",
                    stacklevel=2,
                )
        if "recursive" in discovery:
            cfg.recursive = bool(discovery["recursive"])

    return cfg

def strategy(self, section: str) -> SectionStrategy:
    """In-section strategy for ``section`` (``"keep"`` if unset)."""
    return self.strategies.get(section, "keep")