128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from typing import Callable
|
|
|
|
ReactiveFunctions = dict[tuple[str, str], Callable[..., object]]
|
|
|
|
|
|
@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._reactive: ReactiveFunctions = {}
|
|
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 register_reactive(self, key: tuple[str, str], function: Callable[..., object]) -> None:
|
|
with self._lock:
|
|
self._reactive[key] = function
|
|
|
|
def reactive_function(self, key: tuple[str, str]) -> Callable[..., object] | None:
|
|
with self._lock:
|
|
return self._reactive.get(key)
|
|
|
|
def sources(self) -> set[tuple[str, str]]:
|
|
with self._lock:
|
|
return {
|
|
(page.source_file, page.module) for page in self._pages.values() if page.source_file
|
|
} | {
|
|
(path, function.__module__) for (path, _), function in self._reactive.items()
|
|
}
|
|
|
|
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) -> tuple[dict[str, PageDefinition], ReactiveFunctions]:
|
|
with self._lock:
|
|
return self._pages.copy(), self._reactive.copy()
|
|
|
|
def restore(self, snapshot: tuple[dict[str, PageDefinition], ReactiveFunctions]) -> None:
|
|
with self._lock:
|
|
self._pages, self._reactive = 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
|
|
}
|
|
self._reactive = {
|
|
key: function for key, function in self._reactive.items() if key[0] != target
|
|
}
|
|
|
|
def clear(self) -> None:
|
|
"""Clear all registered pages. Primarily useful in test suites."""
|
|
with self._lock:
|
|
self._pages.clear()
|
|
self._reactive.clear()
|
|
|
|
|
|
registry = Registry()
|