Coverage for greyhorse / river / operator.py: 100%

1"""Operator decorator for marking methods as orchestration primitives. 

2 

3The ``@operator`` decorator turns a class method into a lazy invocation builder. 

4Calling the decorated method returns an ``OperatorInvocation`` instead of executing 

5the method body directly. The invocation can then be configured with ``bind()`` / 

6``bind_resource()``, and materialized into a lazy ``Resource`` via ``prepare(resolver)``. 

7 

8Note that ``prepare()`` itself is lazy — it builds a ``Resource[IO, T]`` program 

9without executing anything. The method body runs only when the Resource is consumed 

10via ``.run()`` or ``with .open() as value:``. 

11 

12Schematic usage:: 

13 

14 ops = DbOps() 

15 inv = ops.publish_pool() # OperatorInvocation (no execution) 

16 program = inv.prepare(resolver) # Resource[IO, T] (still no execution) 

17 result = program.run() # NOW the method body executes 

18""" 

19 

20from __future__ import annotations 

21 

22from collections.abc import Callable 

23from types import MethodType 

24from typing import Any 

25from weakref import WeakKeyDictionary 

26 

27from greyhorse.factory import into_factory 

28 

29from .invocation import OperatorInvocation 

30 

31 

32def operator(method: Callable) -> _OperatorDescriptor: 

33 """Mark a method as an operator — a lazy orchestration primitive. 

34 

35 The decorated method becomes a descriptor. Accessing it on an instance 

36 returns a callable that produces ``OperatorInvocation`` objects. 

37 

38 Uses ``MethodType`` for self-binding and ``WeakKeyDictionary`` to cache 

39 one ``Factory`` per owner instance (no leaks on GC). 

40 

41 Args: 

42 method: The unbound method to decorate. 

43 

44 Returns: 

45 A descriptor that produces ``OperatorInvocation`` builders on access. 

46 

47 Examples: 

48 Schematic (requires a Component with fragments for full execution):: 

49 

50 class MyOps: 

51 @operator 

52 def compute(self, x: int) -> int: 

53 return x * 2 

54 

55 ops = MyOps() 

56 inv = ops.compute(x=5) # OperatorInvocation 

57 program = inv.prepare(comp) # Resource[IO, int] — lazy 

58 result = program.run() # executes the method body 

59 """ 

60 return _OperatorDescriptor(method) 

61 

62 

63class _OperatorDescriptor: 

64 """Descriptor that binds ``@operator``-decorated methods to owner instances.""" 

65 

66 __slots__ = ('_factory_cache', '_method') 

67 

68 def __init__(self, method: Callable) -> None: 

69 self._method = method 

70 self._factory_cache: WeakKeyDictionary = WeakKeyDictionary() 

71 

72 def __set_name__(self, owner: type, name: str) -> None: 

73 pass 

74 

75 def __get__(self, owner_self: Any, owner_cls: type | None = None) -> Any: 

76 if owner_self is None: 

77 return self 

78 

79 try: 

80 if owner_self not in self._factory_cache: 

81 bound = MethodType(self._method, owner_self) 

82 self._factory_cache[owner_self] = into_factory(bound) 

83 factory = self._factory_cache[owner_self] 

84 except TypeError: 

85 bound = MethodType(self._method, owner_self) 

86 factory = into_factory(bound) 

87 

88 def invocation_builder(*args: Any, **kwargs: Any) -> OperatorInvocation: 

89 return OperatorInvocation(factory, *args, **kwargs) 

90 

91 return invocation_builder