agent_search_gateway.providers.academic.registry

 1"""Registration-order-preserving academic provider and OA resolver registries."""
 2
 3from collections.abc import Callable
 4from dataclasses import dataclass
 5from typing import Literal
 6
 7Requirement = Literal["none", "optional", "required"]
 8AcademicProviderFactory = Callable[..., object]
 9OAResolverFactory = Callable[..., object]
10
11
12@dataclass(frozen=True, slots=True)
13class AcademicProviderRegistration:
14    name: str
15    factory: AcademicProviderFactory
16    allowed_config_keys: frozenset[str] = frozenset()
17    authentication: Requirement = "none"
18    contact: Requirement = "none"
19
20
21@dataclass(frozen=True, slots=True)
22class OAResolverRegistration:
23    name: str
24    factory: OAResolverFactory
25    allowed_config_keys: frozenset[str] = frozenset()
26    authentication: Requirement = "none"
27    contact: Requirement = "none"
28
29
30class AcademicProviderRegistry:
31    def __init__(self) -> None:
32        self._registrations: dict[str, AcademicProviderRegistration] = {}
33
34    def register(self, registration: AcademicProviderRegistration) -> None:
35        _validate_registration(registration.name, registration.authentication, registration.contact)
36        if registration.name in self._registrations:
37            raise ValueError(f"academic provider already registered: {registration.name}")
38        self._registrations[registration.name] = registration
39
40    def get(self, name: str) -> AcademicProviderRegistration | None:
41        return self._registrations.get(name)
42
43    def require(self, name: str) -> AcademicProviderRegistration:
44        registration = self.get(name)
45        if registration is None:
46            raise KeyError(name)
47        return registration
48
49    def list_in_registration_order(self) -> tuple[AcademicProviderRegistration, ...]:
50        return tuple(self._registrations.values())
51
52
53class OAResolverRegistry:
54    def __init__(self) -> None:
55        self._registrations: dict[str, OAResolverRegistration] = {}
56
57    def register(self, registration: OAResolverRegistration) -> None:
58        _validate_registration(registration.name, registration.authentication, registration.contact)
59        if registration.name in self._registrations:
60            raise ValueError(f"OA resolver already registered: {registration.name}")
61        self._registrations[registration.name] = registration
62
63    def get(self, name: str) -> OAResolverRegistration | None:
64        return self._registrations.get(name)
65
66    def require(self, name: str) -> OAResolverRegistration:
67        registration = self.get(name)
68        if registration is None:
69            raise KeyError(name)
70        return registration
71
72    def list_in_registration_order(self) -> tuple[OAResolverRegistration, ...]:
73        return tuple(self._registrations.values())
74
75
76def _validate_registration(name: str, authentication: str, contact: str) -> None:
77    if not name.strip():
78        raise ValueError("registration name must be non-empty")
79    valid = {"none", "optional", "required"}
80    if authentication not in valid:
81        raise ValueError(f"invalid authentication requirement: {authentication}")
82    if contact not in valid:
83        raise ValueError(f"invalid contact requirement: {contact}")