mode

AsyncIO Service-based programming.

class mode.Service(*, beacon: mode.utils.types.trees.NodeT = None, loop: asyncio.events.AbstractEventLoop = None)

An asyncio service that can be started/stopped/restarted.

Keyword Arguments:  

• beacon (NodeT) – Beacon used to track services in a graph.
• loop (asyncio.AbstractEventLoop) – Event loop object.

abstract = False

class Diag(service: mode.types.services.ServiceT)

Service diagnostics.

This can be used to track what your service is doing. For example if your service is a Kafka consumer with a background thread that commits the offset every 30 seconds, you may want to see when this happens:

DIAG_COMMITTING = 'committing'

class Consumer(Service):

    @Service.task
    async def _background_commit(self) -> None:
        while not self.should_stop:
            await self.sleep(30.0)
            self.diag.set_flag(DIAG_COMITTING)
            try:
                await self._consumer.commit()
            finally:
                self.diag.unset_flag(DIAG_COMMITTING)

The above code is setting the flag manually, but you can also use a decorator to accomplish the same thing:

@Service.timer(30.0)
async def _background_commit(self) -> None:
    await self.commit()

@Service.transitions_with(DIAG_COMITTING)
async def commit(self) -> None:
    await self._consumer.commit()

set_flag(flag: str) → None

unset_flag(flag: str) → None

wait_for_shutdown = False

Set to True if .stop must wait for the shutdown flag to be set.

shutdown_timeout = 60.0

Time to wait for shutdown flag set before we give up.

restart_count = 0

Current number of times this service instance has been restarted.

mundane_level = 'info'

The log level for mundane info such as starting, stopping, etc. Set this to "debug" for less information.

classmethod from_awaitable(coro: Awaitable, *, name: str = None, **kwargs) → mode.types.services.ServiceT

classmethod task(fun: Callable[Any, Awaitable[None]]) → mode.services.ServiceTask

Decorate function to be used as background task.

Example

>>> class S(Service):
...
...     @Service.task
...     async def background_task(self):
...         while not self.should_stop:
...             await self.sleep(1.0)
...             print('Waking up')

classmethod timer(interval: Union[datetime.timedelta, float, str]) → Callable[Callable[mode.types.services.ServiceT, Awaitable[None]], mode.services.ServiceTask]

Background timer executing every n seconds.

Example

>>> class S(Service):
...
...     @Service.timer(1.0)
...     async def background_timer(self):
...         print('Waking up')

classmethod transitions_to(flag: str) → Callable

Decorate function to set and reset diagnostic flag.

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

Add dependency to other service.

The service will be started/stopped with this service.

add_context(context: ContextManager) → Any

add_future(coro: Awaitable) → _asyncio.Future

Add relationship to asyncio.Future.

The future will be joined when this service is stopped.

on_init() → None

on_init_dependencies() → Iterable[mode.types.services.ServiceT]

Return list of service dependencies for this service.

coroutine add_async_context(context: AsyncContextManager) → Any

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

coroutine crash(reason: BaseException) → None

Crash the service and all child services.

coroutine join_services(services: Sequence[mode.types.services.ServiceT]) → None

logger = <Logger mode.services (WARNING)>

coroutine maybe_start() → None

Start the service, if it has not already been started.

coroutine restart() → None

Restart this service.

service_reset() → None

coroutine sleep(n: Union[datetime.timedelta, float, str]) → None

Sleep for n seconds, or until service stopped.

coroutine start() → None

coroutine stop() → None

Stop the service.

coroutine transition_with(flag: str, fut: Awaitable, *args, **kwargs) → Any

coroutine wait(*coros, timeout: Union[datetime.timedelta, float, str] = None) → mode.services.WaitResult

Wait for coroutines to complete, or until the service stops.

coroutine wait_first(*coros, timeout: Union[datetime.timedelta, float, str] = None) → mode.services.WaitResults

coroutine wait_for_stopped(*coros, timeout: Union[datetime.timedelta, float, str] = None) → bool

coroutine wait_many(coros: Iterable[Union[Generator[[Any, None], Any], Awaitable, asyncio.locks.Event, mode.utils.locks.Event]], *, timeout: Union[datetime.timedelta, float, str] = None) → mode.services.WaitResult

coroutine wait_until_stopped() → None

Wait until the service is signalled to stop.

set_shutdown() → None

Set the shutdown signal.

Notes

If wait_for_shutdown is set, stopping the service will wait for this flag to be set.

started

Return True if the service was started.

crashed

should_stop

Return True if the service must stop.

state

Service state - as a human readable string.

label

Label used for graphs.

shortlabel

Label used for logging.

beacon

Beacon used to track services in a dependency graph.

class mode.BaseSignal(*, name: str = None, owner: Type = None, loop: asyncio.events.AbstractEventLoop = None, default_sender: Any = None, receivers: MutableSet[Any] = None, filter_receivers: MutableMapping[Any, MutableSet[Any]] = None)

Base class for signal/observer pattern.

asdict() → Mapping[str, Any]

clone(**kwargs) → mode.types.signals.BaseSignalT

with_default_sender(sender: Any = None) → mode.types.signals.BaseSignalT

unpack_sender_from_args(*args) → Tuple[T, Tuple[Any, ...]]

connect(fun: Union[Callable[[T, Any, BaseSignalT, Any], None], Callable[[T, Any, BaseSignalT, Any], Awaitable[None]]] = None, **kwargs) → Callable

disconnect(fun: Union[Callable[[T, Any, BaseSignalT, Any], None], Callable[[T, Any, BaseSignalT, Any], Awaitable[None]]], *, weak: bool = False, sender: Any = None) → None

iter_receivers(sender: T_contra) → Iterable[Union[Callable[[T, Any, mode.types.signals.BaseSignalT, Any], None], Callable[[T, Any, mode.types.signals.BaseSignalT, Any], Awaitable[None]]]]

ident

label

class mode.Signal(*, name: str = None, owner: Type = None, loop: asyncio.events.AbstractEventLoop = None, default_sender: Any = None, receivers: MutableSet[Any] = None, filter_receivers: MutableMapping[Any, MutableSet[Any]] = None)

Asynchronous signal (using async def functions).

clone(**kwargs) → mode.types.signals.SignalT

with_default_sender(sender: Any = None) → mode.types.signals.SignalT

coroutine send(*args, **kwargs) → None

class mode.SyncSignal(*, name: str = None, owner: Type = None, loop: asyncio.events.AbstractEventLoop = None, default_sender: Any = None, receivers: MutableSet[Any] = None, filter_receivers: MutableMapping[Any, MutableSet[Any]] = None)

Signal that is synchronous (using regular def functions).

send(*args, **kwargs) → None

clone(**kwargs) → mode.types.signals.SyncSignalT

with_default_sender(sender: Any = None) → mode.types.signals.SyncSignalT

class mode.ForfeitOneForAllSupervisor(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = <class 'mode.exceptions.MaxRestartsExceeded'>, replacement: Callable[[mode.types.services.ServiceT, int], Awaitable[mode.types.services.ServiceT]] = None, **kwargs)

If one service in the group crashes, we give up on all of them.

logger = <Logger mode.supervisors (WARNING)>

coroutine restart_services(services: List[mode.types.services.ServiceT]) → None

class mode.ForfeitOneForOneSupervisor(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = <class 'mode.exceptions.MaxRestartsExceeded'>, replacement: Callable[[mode.types.services.ServiceT, int], Awaitable[mode.types.services.ServiceT]] = None, **kwargs)

Supervisor that if a service crashes, we do not restart it.

logger = <Logger mode.supervisors (WARNING)>

coroutine restart_services(services: List[mode.types.services.ServiceT]) → None

class mode.OneForAllSupervisor(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = <class 'mode.exceptions.MaxRestartsExceeded'>, replacement: Callable[[mode.types.services.ServiceT, int], Awaitable[mode.types.services.ServiceT]] = None, **kwargs)

Supervisor that restarts all services when a service crashes.

logger = <Logger mode.supervisors (WARNING)>

coroutine restart_services(services: List[mode.types.services.ServiceT]) → None

class mode.OneForOneSupervisor(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = <class 'mode.exceptions.MaxRestartsExceeded'>, replacement: Callable[[mode.types.services.ServiceT, int], Awaitable[mode.types.services.ServiceT]] = None, **kwargs)

Supervisor simply restarts any crashed service.

logger = <Logger mode.supervisors (WARNING)>

class mode.SupervisorStrategy(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = <class 'mode.exceptions.MaxRestartsExceeded'>, replacement: Callable[[mode.types.services.ServiceT, int], Awaitable[mode.types.services.ServiceT]] = None, **kwargs)

Base class for all supervisor strategies.

wakeup() → None

add(*services) → None

discard(*services) → None

insert(index: int, service: mode.types.services.ServiceT) → None

service_operational(service: mode.types.services.ServiceT) → bool

logger = <Logger mode.supervisors (WARNING)>

coroutine on_start() → None

Service is starting.

coroutine on_stop() → None

Service is being stopped/restarted.

coroutine restart_service(service: mode.types.services.ServiceT) → None

coroutine restart_services(services: List[mode.types.services.ServiceT]) → None

coroutine run_until_complete() → None

coroutine start_service(service: mode.types.services.ServiceT) → None

coroutine start_services(services: List[mode.types.services.ServiceT]) → None

coroutine stop_services(services: List[mode.types.services.ServiceT]) → None

class mode.CrashingSupervisor(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = <class 'mode.exceptions.MaxRestartsExceeded'>, replacement: Callable[[mode.types.services.ServiceT, int], Awaitable[mode.types.services.ServiceT]] = None, **kwargs)

Supervisor that crashes the whole program.

logger = <Logger mode.supervisors (WARNING)>

wakeup() → None

class mode.ServiceT(*, beacon: mode.utils.types.trees.NodeT = None, loop: asyncio.events.AbstractEventLoop = None)

Abstract type for an asynchronous service that can be started/stopped.

See also

mode.Service.

wait_for_shutdown = False

restart_count = 0

supervisor = None

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

add_context(context: ContextManager) → Any

service_reset() → None

set_shutdown() → None

started

crashed

should_stop

state

label

shortlabel

beacon

coroutine add_async_context(context: AsyncContextManager) → Any

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

coroutine crash(reason: BaseException) → None

coroutine maybe_start() → None

coroutine restart() → None

coroutine start() → None

coroutine stop() → None

coroutine wait_until_stopped() → None

loop

class mode.BaseSignalT(*, name: str = None, owner: Type = None, loop: asyncio.events.AbstractEventLoop = None, default_sender: Any = None, receivers: MutableSet[Any] = None, filter_receivers: MutableMapping[Any, MutableSet[Any]] = None)

Base type for all signals.

clone(**kwargs) → mode.types.signals.BaseSignalT

with_default_sender(sender: Any = None) → mode.types.signals.BaseSignalT

connect(fun: Union[Callable[[T, Any, BaseSignalT, Any], None], Callable[[T, Any, BaseSignalT, Any], Awaitable[None]]], **kwargs) → Callable

disconnect(fun: Union[Callable[[T, Any, BaseSignalT, Any], None], Callable[[T, Any, BaseSignalT, Any], Awaitable[None]]], *, sender: Any = None, weak: bool = True) → None

class mode.SignalT(*, name: str = None, owner: Type = None, loop: asyncio.events.AbstractEventLoop = None, default_sender: Any = None, receivers: MutableSet[Any] = None, filter_receivers: MutableMapping[Any, MutableSet[Any]] = None)

Base class for all async signals (using async def).

clone(**kwargs) → SignalT

with_default_sender(sender: Any = None) → SignalT

coroutine send(sender: T_contra, *args, **kwargs) → None

class mode.SyncSignalT(*, name: str = None, owner: Type = None, loop: asyncio.events.AbstractEventLoop = None, default_sender: Any = None, receivers: MutableSet[Any] = None, filter_receivers: MutableMapping[Any, MutableSet[Any]] = None)

Base class for all synchronous signals (using regular def).

send(sender: T_contra, *args, **kwargs) → None

clone(**kwargs) → SyncSignalT

with_default_sender(sender: Any = None) → SyncSignalT

class mode.SupervisorStrategyT(*services, max_restarts: Union[datetime.timedelta, float, str] = 100.0, over: Union[datetime.timedelta, float, str] = 1.0, raises: Type[BaseException] = None, replacement: Callable[[mode.types.supervisors.ServiceT, int], Awaitable[mode.types.supervisors.ServiceT]] = None, **kwargs)

Base type for all supervisor strategies.

wakeup() → None

add(*services) → None

discard(*services) → None

service_operational(service: mode.types.supervisors.ServiceT) → bool

coroutine restart_service(service: mode.types.supervisors.ServiceT) → None

mode.want_seconds(s: float) → float

Convert Seconds to float.

class mode.flight_recorder(logger: Any, *, timeout: Union[datetime.timedelta, float, str], loop: asyncio.events.AbstractEventLoop = None)

Flight Recorder context for use with with statement.

This is a logging utility to log stuff only when something times out.

For example if you have a background thread that is sometimes hanging:

class RedisCache(mode.Service):

    @mode.timer(1.0)
    def _background_refresh(self) -> None:
        self._users = await self.redis_client.get(USER_KEY)
        self._posts = await self.redis_client.get(POSTS_KEY)

You want to figure out on what line this is hanging, but logging all the time will provide way too much output, and will even change how fast the program runs and that can mask race conditions, so that they never happen.

Use the flight recorder to save the logs and only log when it times out:

logger = mode.get_logger(__name__)

class RedisCache(mode.Service):

    @mode.timer(1.0)
    def _background_refresh(self) -> None:
        with mode.flight_recorder(logger, timeout=10.0) as on_timeout:
            on_timeout.info(f'+redis_client.get({USER_KEY!r})')
            await self.redis_client.get(USER_KEY)
            on_timeout.info(f'-redis_client.get({USER_KEY!r})')

            on_timeout.info(f'+redis_client.get({POSTS_KEY!r})')
            await self.redis_client.get(POSTS_KEY)
            on_timeout.info(f'-redis_client.get({POSTS_KEY!r})')

If the body of this with statement completes before the timeout, the logs are forgotten about and never emitted – if it takes more than ten seconds to complete, we will see these messages in the log:

[2018-04-19 09:43:55,877: WARNING]: Warning: Task timed out!
[2018-04-19 09:43:55,878: WARNING]:
    Please make sure it is hanging before restarting.
[2018-04-19 09:43:55,878: INFO]: [Flight Recorder-1]
    (started at Thu Apr 19 09:43:45 2018) Replaying logs...
[2018-04-19 09:43:55,878: INFO]: [Flight Recorder-1]
    (Thu Apr 19 09:43:45 2018) +redis_client.get('user')
[2018-04-19 09:43:55,878: INFO]: [Flight Recorder-1]
    (Thu Apr 19 09:43:49 2018) -redis_client.get('user')
[2018-04-19 09:43:55,878: INFO]: [Flight Recorder-1]
    (Thu Apr 19 09:43:46 2018) +redis_client.get('posts')
[2018-04-19 09:43:55,878: INFO]: [Flight Recorder-1] -End of log-

Now we know this redis_client.get call can take too long to complete, and should consider adding a timeout to it.

wrap_debug(obj: Any) → mode.utils.logging.Logwrapped

wrap_info(obj: Any) → mode.utils.logging.Logwrapped

wrap_warn(obj: Any) → mode.utils.logging.Logwrapped

wrap_error(obj: Any) → mode.utils.logging.Logwrapped

wrap(severity: int, obj: Any) → mode.utils.logging.Logwrapped

activate() → None

cancel() → None

log(severity: int, message: str, *args, **kwargs) → None

mode.get_logger(name: str) → logging.Logger

Get logger by name.

mode.setup_logging(*, loglevel: Union[str, int] = None, logfile: Union[str, IO] = None, loghandlers: List[logging.StreamHandler] = None, logging_config: Dict = None) → int

Configure logging subsystem.

mode.label(s: Any) → str

Return the name of an object as string.

mode.shortlabel(s: Any) → str

Return the shortened name of an object as string.

class mode.Worker(*services, debug: bool = False, quiet: bool = False, logging_config: Dict = None, loglevel: Union[str, int] = None, logfile: Union[str, IO] = None, redirect_stdouts: bool = True, redirect_stdouts_level: Union[int, str] = None, stdout: IO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>, stderr: IO = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, console_port: int = 50101, loghandlers: List[logging.StreamHandler] = None, blocking_timeout: Union[datetime.timedelta, float, str] = 10.0, loop: asyncio.events.AbstractEventLoop = None, daemon: bool = True, **kwargs)

Start mode service from the command-line.

say(msg: str) → None

Write message to standard out.

carp(msg: str) → None

Write warning to standard err.

on_init_dependencies() → Iterable[mode.types.services.ServiceT]

Return list of service dependencies for this service.

on_setup_root_logger(logger: logging.Logger, level: int) → None

install_signal_handlers() → None

coroutine default_on_first_start() → None

logger = <Logger mode.worker (WARNING)>

coroutine maybe_start_blockdetection() → None

coroutine on_execute() → None

coroutine on_first_start() → None

Service started for the first time in this process.

coroutine start() → None

execute_from_commandline() → NoReturn

on_worker_shutdown() → None

stop_and_shutdown() → None

blocking_detector