Bases: ServiceBase
A service proxy delegates ServiceT methods to a composite service.
Example:
class MyServiceProxy(ServiceProxy):
@cached_property
def _service(self) -> ServiceT:
return ActualService()
Notes
This is used by Faust, and probably useful elsewhere!
The Faust App is created at module-level, and it uses service
proxy to ensure the event loop is not also created just by importing
a module.
Source code in mode/proxy.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118 | class ServiceProxy(ServiceBase):
"""A service proxy delegates ServiceT methods to a composite service.
Example:
```python
class MyServiceProxy(ServiceProxy):
@cached_property
def _service(self) -> ServiceT:
return ActualService()
```
Notes:
This is used by Faust, and probably useful elsewhere!
The Faust App is created at module-level, and it uses service
proxy to ensure the event loop is not also created just by importing
a module.
"""
@property
@abc.abstractmethod
def _service(self) -> ServiceT: # pragma: no cover
...
def add_dependency(self, service: ServiceT) -> ServiceT:
return self._service.add_dependency(service)
async def add_runtime_dependency(self, service: ServiceT) -> ServiceT:
return await self._service.add_runtime_dependency(service)
async def add_async_context(self, context: AsyncContextManager) -> Any:
return await self._service.add_async_context(context)
def add_context(self, context: ContextManager) -> Any:
return self._service.add_context(context)
async def start(self) -> None:
await self._service.start()
async def maybe_start(self) -> bool:
return await self._service.maybe_start()
async def crash(self, reason: BaseException) -> None:
await self._service.crash(reason)
def _crash(self, reason: BaseException) -> None:
self._service._crash(reason)
async def stop(self) -> None:
await self._service.stop()
def service_reset(self) -> None:
self._service.service_reset()
async def restart(self) -> None:
await self._service.restart()
async def wait_until_stopped(self) -> None:
await self._service.wait_until_stopped()
def set_shutdown(self) -> None:
self._service.set_shutdown()
@property
def started(self) -> bool:
return self._service.started
@property
def crashed(self) -> bool:
return self._service.crashed
@property
def should_stop(self) -> bool:
return self._service.should_stop
@property
def state(self) -> str:
return self._service.state
@property
def label(self) -> str:
return type(self).__name__
@property
def shortlabel(self) -> str:
return type(self).__name__
@property
def beacon(self) -> NodeT:
return self._service.beacon
@beacon.setter
def beacon(self, beacon: NodeT) -> None:
self._service.beacon = beacon
@property
def crash_reason(self) -> Optional[BaseException]:
return self._service.crash_reason
@crash_reason.setter
def crash_reason(self, reason: Optional[BaseException]) -> None:
self._service.crash_reason = reason
|