mode.utils.logging

Logging utilities.

mode.utils.logging.FormatterHandler

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.

Example

>>> 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(
            logger=self.logger,
            formatter=self._format_log,
        )

    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:

>>> service.log.info('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 https://gist.github.com/737056.

class mode.utils.logging.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.

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.