77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
__all__ = ('Targets', 'JsonLike', 'Async')
|
|
|
|
import abc
|
|
from typing import Any, Callable, Generic, TypeVar
|
|
|
|
from rainbowadn.instrument import Instrumentation
|
|
|
|
from .responsetype import *
|
|
|
|
|
|
def qualname(t: type) -> str:
|
|
return f'{t.__module__}.{t.__qualname__}'
|
|
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
class Flagful(Generic[T]):
|
|
def __init__(self, value: T, flags: set[object]) -> None:
|
|
self.value = value
|
|
self.flags = flags
|
|
|
|
|
|
class Targets:
|
|
def __init__(self) -> None:
|
|
self.targets: dict[str, Flagful[tuple[Any, str]]] = {}
|
|
self.instrumentations: dict[str, Flagful[Callable[[Any, str], Instrumentation]]] = {}
|
|
self.factories: dict[tuple[str, str], Callable[[], Instrumentation]] = {}
|
|
|
|
def register_target(self, targetname: str, target: Any, methodname: str, /, *flags: object) -> None:
|
|
self.targets[targetname] = Flagful((target, methodname), set(flags))
|
|
print(f'registered target: {targetname}')
|
|
|
|
def register_type(self, target: type, methodname: str, /, *flags: object) -> None:
|
|
self.register_target(f'{qualname(target)}.{methodname}', target, methodname, *flags)
|
|
|
|
def register_instance(self, target: object, methodname: str, /, *flags: object) -> None:
|
|
self.register_target(f'{qualname(target.__class__)}().{methodname}', target, methodname, *flags)
|
|
|
|
def register_instrumentation(
|
|
self,
|
|
instrumentationname: str,
|
|
instrumentation_factory: Callable[[Any, str], Instrumentation],
|
|
/,
|
|
*flags: object,
|
|
) -> None:
|
|
self.instrumentations[instrumentationname] = Flagful(instrumentation_factory, set(flags))
|
|
print(f'registered instrumentation: {instrumentationname}')
|
|
|
|
def get_factory(
|
|
self,
|
|
targetname: str,
|
|
target: Any,
|
|
methodname: str,
|
|
instrumentationname: str,
|
|
instrumentation_factory: Callable[[Any, str], Instrumentation],
|
|
/
|
|
) -> Callable[[], Instrumentation]:
|
|
if (targetname, instrumentationname) not in self.factories:
|
|
flags_required = self.instrumentations[instrumentationname].flags
|
|
flags_present = self.targets[targetname].flags
|
|
if not flags_required.issubset(flags_present):
|
|
raise KeyError('target lacks flags required by instrumentation')
|
|
self.factories[targetname, instrumentationname] = (
|
|
lambda: instrumentation_factory(target, methodname)
|
|
)
|
|
return self.factories[targetname, instrumentationname]
|
|
|
|
|
|
class JsonLike(abc.ABC):
|
|
@abc.abstractmethod
|
|
def json(self) -> ResponseType:
|
|
raise NotImplementedError
|
|
|
|
|
|
Async = object()
|