declarator

addon_toolkit/declarator.py

@@ -0,0 +1,64 @@
+import dataclasses
+import weakref
+
+
+@dataclasses.dataclass
+class Declarator:
+    """add declarative metadata to python functions using decorators"""
+
+    dataclass: type
+    target_fieldname: str
+    static_kwargs: dict | None = None
+
+    # private storage linking a decorated class or function to data gleaned from its decorator
+    __declarations_by_target: weakref.WeakKeyDictionary = dataclasses.field(
+        default_factory=weakref.WeakKeyDictionary,
+    )
+
+    def __post_init__(self):
+        assert any(
+            _field.name == self.target_fieldname
+            for _field in dataclasses.fields(self.dataclass)
+        ), f'expected field "{self.target_fieldname}" on dataclass "{self.dataclass}"'
+
+    def __call__(self, **decorator_kwargs):
+        def _decorator(decorator_target) -> type:
+            self.declare(decorator_target, decorator_kwargs)
+            return decorator_target
+
+        return _decorator
+
+    def with_kwargs(self, **static_kwargs):
+        # note: shared __declarations_by_target
+        return dataclasses.replace(self, static_kwargs=static_kwargs)
+
+    def declare(self, decorator_target, decorator_kwargs: dict):
+        self.__declarations_by_target[decorator_target] = self.dataclass(
+            **decorator_kwargs,
+            **(self.static_kwargs or {}),
+            **{self.target_fieldname: decorator_target},
+        )
+
+    def get_declaration(self, target):
+        try:
+            return self.__declarations_by_target[target]
+        except KeyError:
+            raise ValueError(f"no declaration found for {target}")
+
+
+class ClassDeclarator(Declarator):
+    """add declarative metadata to python classes using decorators"""
+
+    def get_declaration_for_class_or_instance(self, type_or_object: type | object):
+        _cls = (
+            type_or_object if isinstance(type_or_object, type) else type(type_or_object)
+        )
+        return self.get_declaration_for_class(_cls)
+
+    def get_declaration_for_class(self, cls: type):
+        for _cls in cls.__mro__:
+            try:
+                return self.get_declaration(_cls)
+            except ValueError:  # TODO: more helpful exception
+                pass
+        raise ValueError(f"no declaration found for {cls}")

addon_toolkit/interface.py

@@ -2,14 +2,16 @@
 import enum
 import inspect
 import logging
-import weakref
 from typing import (
     Callable,
-    ClassVar,
     Iterator,
 )
 
-from .operation import AddonOperationDeclaration
+from .declarator import ClassDeclarator
+from .operation import (
+    AddonOperationDeclaration,
+    operation_declarator,
+)
 
 
 __all__ = (
@@ -44,40 +46,6 @@ class AddonInterfaceDeclaration:
         default_factory=dict,
     )
 
-    ###
-    # AddonInterfaceDeclaration stores references to declared interface classes
-
-    # private storage linking a class to data gleaned from its decorator
-    __declarations_by_cls: ClassVar[
-        weakref.WeakKeyDictionary[type, "AddonInterfaceDeclaration"]
-    ] = weakref.WeakKeyDictionary()
-
-    @classmethod
-    def declare(cls, capability_enum: type[enum.Enum]):
-        def _cls_decorator(interface_cls: type) -> type:
-            cls.__declarations_by_cls[interface_cls] = AddonInterfaceDeclaration(
-                interface_cls, capability_enum
-            )
-            return interface_cls
-
-        return _cls_decorator
-
-    @classmethod
-    def for_class_or_instance(
-        cls, interface: type | object
-    ) -> "AddonInterfaceDeclaration":
-        _interface_cls = interface if isinstance(interface, type) else type(interface)
-        return cls.for_class(_interface_cls)
-
-    @classmethod
-    def for_class(cls, interface_cls: type) -> "AddonInterfaceDeclaration":
-        for _cls in interface_cls.__mro__:
-            try:
-                return AddonInterfaceDeclaration.__declarations_by_cls[_cls]
-            except KeyError:
-                pass
-        raise ValueError(f"no addon_interface declaration found for {interface_cls}")
-
     ###
     # private methods for populating operations
 
@@ -86,7 +54,7 @@ def __post_init__(self):
 
     def _gather_operations(self):
         for _name, _fn in inspect.getmembers(self.interface_cls, inspect.isfunction):
-            _maybe_op = AddonOperationDeclaration.get_for_function(_fn)
+            _maybe_op = operation_declarator.get_declaration(_fn)
             if _maybe_op is not None:
                 self._add_operation(_name, _maybe_op)
 
@@ -101,8 +69,11 @@ def _add_operation(self, method_name: str, operation: AddonOperationDeclaration)
         ).add(operation)
 
 
-# meant for use as decorator on a class, `@addon_interface(MyCapabilitiesEnum)`
-addon_interface = AddonInterfaceDeclaration.declare
+# the class decorator itself
+addon_interface = ClassDeclarator(
+    dataclass=AddonInterfaceDeclaration,
+    target_fieldname="interface_cls",
+)
 
 
 @dataclasses.dataclass(frozen=True)
@@ -113,15 +84,18 @@ class AddonOperationImplementation:
     operation: AddonOperationDeclaration
 
     def __post_init__(self):
-        _interface_cls_fn = getattr(self.interface.interface_cls, self.method_name)
-        if self.implementation_fn is _interface_cls_fn:
+        if self.implementation_fn is self.interface_fn:  # may raise NotImplementedError
             raise NotImplementedError(  # TODO: helpful exception type
                 f"operation '{self.operation}' not implemented by {self.implementation_cls}"
             )
 
     @property
     def interface(self) -> AddonInterfaceDeclaration:
-        return AddonInterfaceDeclaration.for_class(self.implementation_cls)
+        return addon_interface.get_declaration_for_class(self.implementation_cls)
+
+    @property
+    def interface_fn(self) -> Callable:
+        return getattr(self.interface.interface_cls, self.method_name)
 
     @property
     def method_name(self) -> str:
@@ -147,7 +121,7 @@ def get_callable_for(self, addon_instance: object) -> Callable:
 def get_operation_declarations(
     interface: type | object, capability: enum.Enum | None = None
 ) -> Iterator[AddonOperationDeclaration]:
-    _interface_dec = AddonInterfaceDeclaration.for_class_or_instance(interface)
+    _interface_dec = addon_interface.get_declaration_for_class_or_instance(interface)
     if capability is None:
         yield from _interface_dec.method_name_by_op.keys()
     else:

addon_toolkit/operation.py

@@ -1,17 +1,14 @@
 import dataclasses
 import enum
-import inspect
-import weakref
-from typing import (
-    Callable,
-    ClassVar,
-    Optional,
-)
+from typing import Callable
+
+from .declarator import Declarator
 
 
 __all__ = (
     "AddonOperationDeclaration",
     "AddonOperationType",
+    "operation_declarator",
     "proxy_operation",
     "redirect_operation",
 )
@@ -34,32 +31,6 @@ class AddonOperationDeclaration:
     capability: enum.Enum
     operation_fn: Callable
 
-    ###
-    # AddonOperationDeclaration stores references to declared operations
-
-    # private storage linking a function to data gleaned from its decorator
-    __operations_by_fn: ClassVar[
-        weakref.WeakKeyDictionary[Callable, "AddonOperationDeclaration"]
-    ] = weakref.WeakKeyDictionary()
-
-    @staticmethod
-    def declared(
-        operation_fn: Callable,
-        capability: enum.Enum,
-        operation_type: AddonOperationType,
-    ):
-        AddonOperationDeclaration.__operations_by_fn[
-            operation_fn
-        ] = AddonOperationDeclaration(
-            operation_type=operation_type,
-            capability=capability,
-            operation_fn=operation_fn,
-        )
-
-    @staticmethod
-    def get_for_function(fn: Callable) -> Optional["AddonOperationDeclaration"]:
-        return AddonOperationDeclaration.__operations_by_fn.get(fn)
-
     ###
     # instance methods
 
@@ -68,29 +39,24 @@ def docstring(self) -> str | None:
         # TODO: consider docstring param on operation decorators, allow overriding __doc__
         return self.operation_fn.__doc__
 
+    @classmethod
+    def for_function(self, fn: Callable) -> "AddonOperationDeclaration":
+        return operation_declarator.get_declaration(fn)
 
-def redirect_operation(capability: enum.Enum):
-    def _redirect_operation_decorator(fn: Callable) -> Callable:
-        # decorator for operations that may be performed by a client request
-        # (e.g. redirect to waterbutler)
-        assert inspect.isfunction(fn)  # TODO: inspect function params
-        assert not inspect.isawaitable(fn)
-        # TODO: helpful error messaging for implementers
-        AddonOperationDeclaration.declared(fn, capability, AddonOperationType.REDIRECT)
-        return fn
-
-    return _redirect_operation_decorator
 
+# decorator for operations (used by operation_type-specific decorators below)
+operation_declarator = Declarator(
+    dataclass=AddonOperationDeclaration,
+    target_fieldname="operation_fn",
+)
 
-def proxy_operation(capability: enum.Enum):
-    def _proxy_operation_decorator(fn: Callable) -> Callable:
-        # decorator for operations that require fetching data from elsewhere,
-        # but make no changes (e.g. get a metadata description of an item,
-        # list items in a given folder)
-        # TODO: assert inspect.isasyncgenfunction(fn)  # generate rdf triples?
-        # TODO: assert based on `inspect.signature(fn).parameters`
-        # TODO: assert based on return value?
-        AddonOperationDeclaration.declared(fn, capability, AddonOperationType.PROXY)
-        return fn
+# decorator for operations that may be performed by a client request (e.g. redirect to waterbutler)
+redirect_operation = operation_declarator.with_kwargs(
+    operation_type=AddonOperationType.REDIRECT,
+)
 
-    return _proxy_operation_decorator
+# decorator for operations that require fetching data from elsewhere, but make no changes
+# (e.g. get a metadata description of an item, list items in a given folder)
+proxy_operation = operation_declarator.with_kwargs(
+    operation_type=AddonOperationType.PROXY,
+)