SolitaryThinker
4e331eb7ff
Mechanical conversion of every relative import under v2/ to an absolute v2.* path (from .x / ..x / ...x -> from v2.<pkg>.x) so imports are unambiguous, grep-able, and stable when code is copied/moved between entrypoints (VideoGenerator, CLI, server). Surgical prefix-only rewrite: only the 'from <dots><module>' prefix changed — import names, parentheses, multi-line formatting, comments, and ordering are byte-for-byte preserved (no collapsing, no reorder, no unrelated reformatting). - 471 imports across 125 files; v2/tests/ was already absolute (untouched). - Validated: all 125 files compile, every 'from v2.* import' target resolves to a real module/package, zero relative imports remain (full sweep), CPU mini suite green (202 passed, 2 skipped).
38 lines
1.5 KiB
Python
38 lines
1.5 KiB
Python
"""Plugin registry + trust boundary (design_v3 §11).
|
|
|
|
> Plugins are enabled at deploy scope only (never a per-request ``plugins=[...]`` field
|
|
> that would wire third-party code selection into the public API); requests only
|
|
> *parameterize* pre-enabled plugins through validated schemas.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable
|
|
|
|
from v2.extend.base import InterceptorChain, Observer
|
|
|
|
|
|
class PluginRegistry:
|
|
def __init__(self) -> None:
|
|
self._interceptors: dict[str, Callable[..., Any]] = {}
|
|
self._observers: dict[str, Callable[..., Observer]] = {}
|
|
self._enabled: set[str] = set() # deploy-scope enablement
|
|
|
|
def register_interceptor(self, plugin_id: str, factory: Callable[..., Any]) -> None:
|
|
self._interceptors[plugin_id] = factory
|
|
|
|
def register_observer(self, plugin_id: str, factory: Callable[..., Observer]) -> None:
|
|
self._observers[plugin_id] = factory
|
|
|
|
def enable(self, plugin_id: str) -> None:
|
|
if plugin_id not in self._interceptors and plugin_id not in self._observers:
|
|
raise KeyError(f"plugin {plugin_id!r} not registered")
|
|
self._enabled.add(plugin_id)
|
|
|
|
def build_chain(self, params: dict[str, dict] | None = None, *, exact_mode: bool = False) -> InterceptorChain:
|
|
params = params or {}
|
|
chain = []
|
|
for pid in self._enabled:
|
|
if pid in self._interceptors:
|
|
chain.append(self._interceptors[pid](**params.get(pid, {})))
|
|
return InterceptorChain(chain, exact_mode=exact_mode)
|