105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from typing import Callable
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PageDefinition:
|
|
number: str
|
|
title: str
|
|
render: Callable[[], object]
|
|
module: str
|
|
source_file: str | None
|
|
|
|
|
|
class Registry:
|
|
def __init__(self) -> None:
|
|
self._pages: dict[str, PageDefinition] = {}
|
|
self._lock = RLock()
|
|
|
|
def register(self, page: PageDefinition) -> None:
|
|
with self._lock:
|
|
previous = self._pages.get(page.number)
|
|
if previous is not None and previous.render is not page.render:
|
|
raise ValueError(f"Page number {page.number!r} is already registered")
|
|
self._pages[page.number] = page
|
|
|
|
def pages(self) -> list[PageDefinition]:
|
|
with self._lock:
|
|
return list(self._pages.values())
|
|
|
|
def resolve_page(
|
|
self,
|
|
target: str | int | float | Callable[..., object],
|
|
*,
|
|
section: str | int | float | None = None,
|
|
) -> PageDefinition:
|
|
"""Resolve a page by function object, number, or title within an optional section."""
|
|
with self._lock:
|
|
pages = list(self._pages.values())
|
|
|
|
if callable(target):
|
|
matches = [page for page in pages if page.render is target]
|
|
if not matches:
|
|
name = getattr(target, "__name__", repr(target))
|
|
raise ValueError(f"No @Page is registered for function {name}")
|
|
return matches[0]
|
|
|
|
value = str(target).strip()
|
|
if not value:
|
|
raise ValueError("Local link target cannot be empty")
|
|
normalised_number = value.strip(".")
|
|
by_number = [page for page in pages if page.number == normalised_number]
|
|
if by_number:
|
|
return by_number[0]
|
|
|
|
section_number = str(section).strip().strip(".") if section is not None else None
|
|
if section_number == "":
|
|
raise ValueError("Local link section cannot be empty")
|
|
candidates = [
|
|
page
|
|
for page in pages
|
|
if page.title.casefold() == value.casefold()
|
|
and (
|
|
section_number is None
|
|
or page.number == section_number
|
|
or page.number.startswith(section_number + ".")
|
|
)
|
|
]
|
|
if not candidates:
|
|
location = f" inside section {section_number!r}" if section_number else ""
|
|
raise ValueError(f"No page titled {value!r}{location}")
|
|
if len(candidates) > 1:
|
|
numbers = ", ".join(page.number for page in candidates)
|
|
raise ValueError(
|
|
f"Page title {value!r} is ambiguous ({numbers}); provide section= or a page number"
|
|
)
|
|
return candidates[0]
|
|
|
|
def snapshot(self) -> dict[str, PageDefinition]:
|
|
with self._lock:
|
|
return dict(self._pages)
|
|
|
|
def restore(self, snapshot: dict[str, PageDefinition]) -> None:
|
|
with self._lock:
|
|
self._pages = dict(snapshot)
|
|
|
|
def remove_source(self, source_file: str | Path) -> None:
|
|
target = str(Path(source_file).resolve())
|
|
with self._lock:
|
|
self._pages = {
|
|
key: page
|
|
for key, page in self._pages.items()
|
|
if not page.source_file or str(Path(page.source_file).resolve()) != target
|
|
}
|
|
|
|
def clear(self) -> None:
|
|
"""Clear all registered pages. Primarily useful in test suites."""
|
|
with self._lock:
|
|
self._pages.clear()
|
|
|
|
|
|
registry = Registry()
|