Prototype

Author: dandavison

.python-version

@@ -1 +1 @@
-3.13
+3.9

pyproject.toml

@@ -6,9 +6,19 @@ readme = "README.md"
 authors = [
     { name = "Dan Davison", email = "dandavison7@gmail.com" }
 ]
-requires-python = ">=3.13"
-dependencies = []
+requires-python = ">=3.9"
+dependencies = [
+    "pyright==1.1.394",
+    "typing-extensions>=4.12.2",
+]
 
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/nexusrpc"]
+
+
+[tool.ruff]
+extend-ignore = ["E741"]  # Allow single-letter variable names like I, O

src/nexus_rpc/__init__.py

@@ -0,0 +1,304 @@
+import inspect
+import types
+from dataclasses import dataclass, field
+from enum import Enum
+from functools import wraps
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Generic,
+    Optional,
+    Protocol,
+    Type,
+    TypeVar,
+    Union,
+    get_origin,
+)
+
+from typing_extensions import ParamSpec
+
+import nexusrpc.interface
+
+# TODO(dan)
+_ServiceImpl = Any
+_Params = ParamSpec("_Params")
+I = TypeVar("I", contravariant=True)
+O = TypeVar("O", covariant=True)
+S = TypeVar("S", bound=_ServiceImpl)
+ST = TypeVar("ST", bound=Type[_ServiceImpl])
+
+
+@dataclass
+class Link:
+    """
+    Link contains a URL and a Type that can be used to decode the URL.
+    Links can contain any arbitrary information as a percent-encoded URL.
+    It can be used to pass information about the caller to the handler, or vice versa.
+    """
+
+    # The URL must be percent-encoded.
+    url: str
+    # Can describe an actual data type for decoding the URL. Valid chars: alphanumeric, '_', '.',
+    # '/'
+    type: str
+
+
+@dataclass
+class StartOperationOptions:
+    """Options passed by the Nexus caller when starting an operation."""
+
+    headers: dict[str, str] = field(default_factory=dict)
+
+    # A callback URL is required to deliver the completion of an async operation. This URL should be
+    # called by a handler upon completion if the started operation is async.
+    callback_url: Optional[str] = None
+
+    # Optional header fields set by the caller to be attached to the callback request when an
+    # asynchronous operation completes.
+    callback_header: dict[str, str] = field(default_factory=dict)
+
+    # Request ID that may be used by the server handler to dedupe a start request.
+    # By default a v4 UUID will be generated by the client.
+    request_id: Optional[str] = None
+
+    # Links contain arbitrary caller information. Handlers may use these links as
+    # metadata on resources associated with an operation.
+    links: list[Link] = field(default_factory=list)
+
+
+@dataclass
+class CancelOperationOptions:
+    """Options passed by the Nexus caller when cancelling an operation."""
+
+    headers: dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class FetchOperationInfoOptions:
+    """Options passed by the Nexus caller when fetching information about an operation."""
+
+    headers: dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class FetchOperationResultOptions:
+    """Options passed by the Nexus caller when fetching the result of an operation."""
+
+    headers: dict[str, str] = field(default_factory=dict)
+
+
+class OperationState(Enum):
+    """
+    Describes the current state of an operation.
+    """
+
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELED = "canceled"
+    RUNNING = "running"
+
+
+@dataclass
+class OperationInfo:
+    """
+    Information about an operation.
+    """
+
+    # Token identifying the operation (returned on operation start).
+    token: str
+
+    # The operation's status.
+    status: OperationState
+
+
+@dataclass
+class AsyncOperationResult:
+    token: str
+
+
+# TODO(dan): TBD what we'll call this. It's public so should not have an
+# underscore prefix.
+@dataclass
+class _NexusServiceDefinition:
+    name: str
+    operation_factories: dict[str, Callable[[_ServiceImpl], "Operation[Any, Any]"]]
+
+    # TODO(dan): TBD what the name of this look method should be.
+    @classmethod
+    def from_implementation(
+        cls, implementation: Type[_ServiceImpl]
+    ) -> "_NexusServiceDefinition":
+        """
+        Retrieve the service definition that was set by the decorator on a service implementation class.
+        """
+        if defn := getattr(implementation, "__nexus_service__", None):
+            return defn
+        raise ValueError(
+            f"Service implementation {implementation} does not have a service definition. "
+            f"Use the @nexusrpc.handler.service decorator on your class to define a Nexus service implementation."
+        )
+
+
+@dataclass
+class _NexusOperationDefinition:
+    name: str
+
+
+@dataclass
+class service(Generic[ST]):
+    """Decorator that marks a class as a Nexus service implementation.
+
+    Args:
+        interface: The service interface that the service implements.
+        name: The name of the  service. If not provided, the class name will be used.
+
+    Example:
+        ```python
+        @nexusrpc.handler.service(MyServiceInterface, name="my-service")
+        class MyService:
+            ...
+        ```
+    """
+
+    interface: Type[Any]
+    name: Optional[str] = None
+
+    def __call__(self, cls: ST) -> ST:
+        cls.__nexus_service__ = _NexusServiceDefinition(
+            # The name by which the service is addressed in Nexus requests is
+            # the name of the interface, not that of the implementation class.
+            name=self.interface.__name__,
+            operation_factories=self._get_operation_factories(cls, self.interface),
+        )
+        return cls
+
+    @staticmethod
+    def _get_operation_factories(
+        service_cls: ST, interface: Type[Any]
+    ) -> dict[str, Callable[[_ServiceImpl], "Operation[Any, Any]"]]:
+        """
+        Get the operation factories from the class.
+        """
+        # TODO(dan): gather all errors, check type parameters, type variance, etc
+        interface_op_names = set()
+        for name, _type in interface.__annotations__.items():
+            # TODO(dan): use get_args to check type parameters
+            if get_origin(_type) == nexusrpc.interface.NexusOperation:
+                interface_op_names.add(name)
+
+        op_factories = {
+            name: method
+            for name, method in inspect.getmembers(service_cls, inspect.isfunction)
+            if hasattr(method, "__nexus_operation__") and name in interface_op_names
+        }
+        if len(op_factories) < len(interface_op_names):
+            raise ValueError(
+                f"Service {service_cls} does not implement all operations in interface {interface}. "
+                f"Missing operations: {interface_op_names - op_factories.keys()}"
+            )
+        return op_factories
+
+
+class Operation(Protocol, Generic[I, O]):
+    """
+    Interface that must be implemented by an operation in a Nexus service implementation.
+    """
+
+    # start either returns the result synchronously, or returns an operation token. Which path is
+    # taken may be decided at operation handling time.
+    async def start(
+        self, input: I, options: StartOperationOptions
+    ) -> Union[O, AsyncOperationResult]: ...
+
+    async def fetch_info(
+        self, token: str, options: FetchOperationInfoOptions
+    ) -> OperationInfo: ...
+
+    async def fetch_result(
+        self, token: str, options: FetchOperationResultOptions
+    ) -> O: ...
+
+    async def cancel(self, token: str, options: CancelOperationOptions) -> None: ...
+
+
+_OpFactory = Callable[[_ServiceImpl], Operation[Any, Any]]
+F = TypeVar("F", bound=_OpFactory)
+
+
+# TODO(dan): This is following workflow.defn but check that invalid decorator
+# usage is prevented by this implementation style.
+def operation(
+    method: Optional[F] = None,
+    *,
+    name: Optional[str] = None,
+) -> Union[F, Callable[[F], F]]:
+    """
+    Decorator that marks a method as an operation in a Nexus service implementation.
+
+    Args:
+        method: The method to decorate.
+        name: The name of the operation. If not provided, the method name will be used.
+
+    Examples:
+        ```
+        @nexusrpc.handler.operation
+        def my_operation(self) -> Operation[MyInput, MyOutput]:
+            ...
+        ```
+
+        ```
+        @nexusrpc.handler.operation(name="my-operation")
+        def my_operation(self) -> Operation[MyInput, MyOutput]:
+            ...
+        ```
+    """
+
+    def decorator(method: F) -> F:
+        method.__nexus_operation__ = _NexusOperationDefinition(
+            name=name or method.__name__
+        )
+        return method
+
+    if method is None:
+        return decorator
+
+    return decorator(method)
+
+
+# abc? require start?
+class AbstractOperation(Operation[I, O]):
+    async def start(self, input: I, options: StartOperationOptions) -> O:
+        raise NotImplementedError
+
+    async def fetch_info(
+        self, token: str, options: FetchOperationInfoOptions
+    ) -> OperationInfo:
+        raise NotImplementedError
+
+    async def fetch_result(self, token: str, options: FetchOperationResultOptions) -> O:
+        raise NotImplementedError
+
+    async def cancel(self, token: str, options: CancelOperationOptions) -> None:
+        raise NotImplementedError
+
+
+# TODO(dan): support overriding op name
+def sync_operation(
+    start_method: Callable[[S, I, StartOperationOptions], Awaitable[O]],
+) -> Callable[[S], AbstractOperation[I, O]]:
+    def factory(service: S) -> AbstractOperation[I, O]:
+        # A start method defined in this way was written by the user as a method
+        # on a service. We convert it into a method on an operation, but the
+        # method will not access the operation.
+        @wraps(start_method)
+        async def start(_, input: I, options: StartOperationOptions) -> O:
+            return await start_method(service, input, options)
+
+        op = AbstractOperation()
+        op.start = types.MethodType(start, op)
+        return op
+
+    factory.__nexus_operation__ = _NexusOperationDefinition(name=start_method.__name__)
+
+    return factory

src/nexus_rpc/py.typed src/nexusrpc/__init__.pysrc/nexus_rpc/py.typed renamed to src/nexusrpc/__init__.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass
+from typing import Generic, Type, TypeVar
+
+I = TypeVar("I", contravariant=True)
+O = TypeVar("O", covariant=True)
+T = TypeVar("T")
+
+
+class NexusService:
+    pass
+
+
+@dataclass
+class NexusOperation(Generic[I, O]):
+    name: str
+
+
+# TODO(dan): support assigning invalid Python identifier as interface name
+# The name of the interface is the name used to refer to the service in Nexus requests.
+def nexus_service(interface: Type[T]) -> Type[T]:
+    """
+    Decorator that creates a Nexus service interface from an interface protocol type.
+
+    Example:
+        ```python
+        @nexus_service
+        class MyService:
+            ...
+        ```
+    """
+    # TODO
+    data = {name: NexusOperation(name) for name in interface.__annotations__}
+
+    return type(f"{interface.__name__}", (interface,), data)  # type: ignore