Logging utilities.


alias of typing.Callable

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

Get logger by name.

class mode.utils.logging.LogSeverityMixin

Mixin class that delegates standard logging methods to logger.

The class that mixes in this class must define the log method.


>>> class Foo(LogSeverityMixin):
...    logger = get_logger('foo')
...    def log(self,
...            severity: int,
...            message: str,
...            *args: Any, **kwargs: Any) -> None:
...        return self.logger.log(severity, message, *args, **kwargs)
dev(message: str, *args: Any, **kwargs: Any) → None
debug(message: str, *args: Any, **kwargs: Any) → None
info(message: str, *args: Any, **kwargs: Any) → None
warn(message: str, *args: Any, **kwargs: Any) → None
warning(message: str, *args: Any, **kwargs: Any) → None
error(message: str, *args: Any, **kwargs: Any) → None
crit(message: str, *args: Any, **kwargs: Any) → None
critical(message: str, *args: Any, **kwargs: Any) → None
exception(message: str, *args: Any, **kwargs: Any) → None
class mode.utils.logging.CompositeLogger(logger: logging.Logger, formatter: Callable[..., str] = None)

Composite logger for classes.

The class can be used as both mixin and composite, and may also define a .formatter attribute which will reformat any log messages sent.

Service uses this to add logging methods:

class Service(ServiceT):

    log: CompositeLogger

    def __init__(self):
        self.log = CompositeLogger(

    def _format_log(self, severity: int, message: str,
                    *args: Any, **kwargs: Any) -> str:
        return (f'[^{"-" * (self.beacon.depth - 1)}'
                f'{self.shortlabel}]: {message}')

This means those defining a service may also use it to log:

>>>'Something happened')

and when logging additional information about the service is automatically included.

logger: Logger = None
log(severity: int, message: str, *args: Any, **kwargs: Any) → None
format(severity: int, message: str, *args: Any, **kwargs: Any) → str
mode.utils.logging.formatter(fun: Callable[Any, Any]) → Callable[Any, Any]

Register formatter for logging positional args.

class mode.utils.logging.ExtensionFormatter(stream: IO = None, **kwargs: Any)

Formatter that can register callbacks to format args.

Extends colorlog.

format(record: logging.LogRecord) → str

Format a message from a record object.

mode.utils.logging.level_name(loglevel: int) → str

Convert log level to number.

mode.utils.logging.level_number(loglevel: int) → int

Convert log level number to name.

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

Configure logging subsystem.

class mode.utils.logging.Logwrapped(obj: Any, logger: Any = None, severity: Union[int, str] = None, ident: str = '')

Wrap all object methods, to log on call.

obj: Any = None
logger: Any = None
severity: int = None
ident: str = None
mode.utils.logging.cry(file: IO, *, sep1: str = '=', sep2: str = '-', sep3: str = '~', seplen: int = 49) → None

Return stack-trace of all active threads.

See also

Taken from

class mode.utils.logging.flight_recorder(logger: Any, *, timeout: Union[datetime.timedelta, float, str], loop: = 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):

    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):

    def _background_refresh(self) -> None:
        with mode.flight_recorder(logger, timeout=10.0) as on_timeout:
            await self.redis_client.get(USER_KEY)

            await self.redis_client.get(POSTS_KEY)

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.

logger: Any = None
timeout: float = None
loop: asyncio.AbstractEventLoop = None
started_at_date: Optional[str] = None
enabled_by: Optional[asyncio.Task] = None
extra_context: Dict[str, Any] = None
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: Any, **kwargs: Any) → None
blush() → None
flush_logs(ident: str = None) → None
class mode.utils.logging.FileLogProxy(logger: logging.Logger, *, severity: Union[int, str] = None)

File-like object that forwards data to logger.

severity: int = 30
write(s: AnyStr) → int
writelines(lines: Iterable[str]) → None
property buffer
property encoding
property errors
line_buffering() → bool
property newlines
flush() → None
property mode
property name
close() → None
property closed
fileno() → int
isatty() → bool
read(n: int = -1) → AnyStr
readable() → bool
readline(limit: int = -1) → AnyStr
readlines(hint: int = -1) → List[AnyStr]
seek(offset: int, whence: int = 0) → int
seekable() → bool
tell() → int
truncate(size: int = None) → int
writable() → bool
mode.utils.logging.redirect_stdouts(logger: logging.Logger = <Logger mode.redirect (WARNING)>, *, severity: Union[int, str] = None, stdout: bool = True, stderr: bool = True) → Iterator[mode.utils.logging.FileLogProxy]

Redirect sys.stdout and sys.stdout to logger.