log

class naz.log.SimpleLogger(logger_name, level=20, log_metadata=None, handler=None, render_as_json=True)[source]

Bases: logging.Logger

It implements a structured logger that renders logs as either json(default) or python dictionary.

example usage:

logger = SimpleLogger("myLogger")
logger.log(logging.INFO,
           {"event": "web_request", "url": "https://www.google.com/"})
__init__(logger_name, level=20, log_metadata=None, handler=None, render_as_json=True)[source]
Parameters

  • logger_name (str) – name of the logger. it should be unique per logger.

  • level (Union[str, int]) – the level at which to log

  • log_metadata (Union[None, dict]) – metadata that will be included in all log statements

  • handler (Union[None, Handler]) – python logging handler to be attached to this logger. By default, logging.StreamHandler is used.

  • render_as_json – whether to render logs as json or dictionary. By default it renders as json.

Return type

None

log(level, msg, *args, **kwargs)[source]

Log ‘msg % args’ with the integer severity ‘level’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.log(level, “We have a %s”, “mysterious problem”, exc_info=1)

class naz.log.BreachHandler(flushLevel=30, capacity=1000, target=None, flushOnClose=False, heartbeatInterval=None, targetLevel='DEBUG')[source]

Bases: logging.handlers.MemoryHandler

This is an implementation of logging.Handler that puts logs in an in-memory ring buffer. When a trigger condition(eg a certain log level) is met; then all the logs in the buffer are flushed into a given stream(file, stdout etc)

It is a bit like logging.handlers.MemoryHandler except that it does not flush when the ring-buffer capacity is met but only when/if the trigger is met.

It is inspired by the article Triggering Diagnostic Logging on Exception

example usage:

import naz, logging

_handler = naz.log.BreachHandler()
logger = naz.log.SimpleLogger("aha", handler=_handler, log_metadata={"id": "123"})

logger.log(logging.INFO, {"name": "Jayz"})
logger.log(logging.ERROR, {"msg": "Houston, we got 99 problems."})

# or alternatively, to use it with python stdlib logger
logger = logging.getLogger("my-logger")
handler = naz.log.BreachHandler()
formatter = logging.Formatter("%(message)s")
handler.setFormatter(formatter)
handler.setLevel("DEBUG")
if not logger.handlers:
    logger.addHandler(handler)
logger.setLevel("DEBUG")

logger.info("I did records for Tweet before y'all could even tweet - Dr. Missy Elliot")
logger.error("damn")
__init__(flushLevel=30, capacity=1000, target=None, flushOnClose=False, heartbeatInterval=None, targetLevel='DEBUG')[source]
Parameters

  • flushLevel (int) – the log level that will trigger this handler to flush logs to target

  • capacity (int) – the maximum number of log records to store in the ring buffer

  • target (Union[None, Handler]) – the log handler that will be used.

  • flushOnClose (bool) – whether to flush the buffer when the handler is closed even if the flush level hasn’t been exceeded

  • heartbeatInterval (Union[None, float]) – can be a float or None. If it is a float, then a heartbeat log record will be emitted every heartbeatInterval seconds. If it is None(the default), then no heartbeat log record is emitted. If you do decide to set it, a good value is at least 1800(ie 30 minutes).

  • targetLevel (str) – the log level to be applied/set to target

Return type

None

emit(record)[source]

Emit a record. Append the record. If shouldFlush() tells us to, call flush() to process the buffer.

Implementation is taken from logging.handlers.MemoryHandler

Return type

None

shouldFlush(record)[source]

Check for record at the flushLevel or higher. Implementation is mostly taken from logging.handlers.MemoryHandler

Return type

bool