mode.proxy

Proxy to service.

Works like a service, but delegates to underlying service object.

class mode.proxy.ServiceProxy(*, loop: asyncio.events.AbstractEventLoop = None)

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.

add_dependency(service: mode.types.services.ServiceT) → mode.types.services.ServiceT

async add_runtime_dependency(service: mode.types.services.ServiceT) → mode.types.services.ServiceT

async add_async_context(context: AsyncContextManager) → Any

add_context(context: ContextManager) → Any

async start() → None

async maybe_start() → bool

async crash(reason: BaseException) → None

async stop() → None

service_reset() → None

async restart() → None

async wait_until_stopped() → None

set_shutdown() → None

property started

property crashed

property should_stop

property state

property label

property shortlabel

property beacon

abstract: ClassVar[bool] = False

log = None

logger: logging.Logger = <Logger mode.proxy (WARNING)>

property crash_reason