Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init_accepts_overriding_levels(self, styles, padded):
"""
Stdlib levels are rendered aligned, in brackets, and color coded.
"""
my_styles = dev.ConsoleRenderer.get_default_level_styles(
colors=dev._has_colorama
)
my_styles["MY_OH_MY"] = my_styles["critical"]
cr = dev.ConsoleRenderer(
colors=dev._has_colorama, level_styles=my_styles
)
# this would blow up if the level_styles override failed
rv = cr(
None, None, {"event": "test", "level": "MY_OH_MY", "foo": "bar"}
)
# fmt: off
assert (
"[" + dev.RED + styles.bright +
dev._pad("MY_OH_MY", cr._longest_level) +
styles.reset + "] " +
padded +
styles.kv_key + "foo" + styles.reset + "=" +
styles.kv_value + "bar" + styles.reset
EMAIL_PORT = 1025
# Your stuff...
# ------------------------------------------------------------------------------
LOGGING = {
"version": 1,
"disable_existing_loggers": True,
"formatters": {
"plain": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer(colors=False),
},
"colored": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer(colors=True),
},
},
"filters": {},
"handlers": {
"structured_stream": {"class": "logging.StreamHandler", "formatter": "colored"},
"structured_file": {
"class": "logging.handlers.WatchedFileHandler",
"filename": "test.log",
"formatter": "plain",
},
},
"loggers": {"": {"handlers": ["structured_stream"], "level": "INFO"}},
}
structlog.configure(
processors=[
def configure_logger(logger, level):
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.format_exc_info,
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
logger_factory=structlog.stdlib.LoggerFactory(),
)
formatter = structlog.stdlib.ProcessorFormatter(
processor=structlog.dev.ConsoleRenderer()
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(level)
def get_processors(*extra_processors, pretty: bool = True):
processors = [
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt='iso'),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
*extra_processors,
]
if pretty:
renderer = structlog.dev.ConsoleRenderer(pad_event=40)
else:
renderer = structlog.processors.JSONRenderer(rapidjson.dumps)
processors.append(renderer)
return processors
def setup_logging(quiet=False, verbose=False, json=False):
# verbose supersedes quiet
level = logging.WARNING if quiet else logging.INFO
level = logging.DEBUG if verbose else level
# root logger
root = logging.getLogger(LOGGER_NAME)
root.setLevel(level)
root.addHandler(logging.StreamHandler(sys.stdout))
# renderer
renderer = (
structlog.processors.JSONRenderer()
if json
else structlog.dev.ConsoleRenderer()
)
# https://structlog.readthedocs.io/en/stable/development.html
structlog.configure(
processors=DEFAULT_PROCESSORS + [renderer],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
clutter for local development.
"""
debug = os.environ.get('DEBUG', 'false') != 'false'
logging.basicConfig(level='DEBUG' if debug else 'INFO',
stream=sys.stdout,
format="%(message)s")
if os.getenv('LOG_MODE', 'JSON') == 'LOCAL':
chain = [
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S.%f"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.dev.ConsoleRenderer()
]
else:
chain = [
LogEntryProcessor.add_app_info,
LogEntryProcessor.add_logger_name,
LogEntryProcessor.add_timestamp,
LogEntryProcessor.censor_password,
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
LogEntryProcessor.cleanup_keynames,
structlog.processors.JSONRenderer()
]
structlog.configure_once(
processors=chain,
context_class=dict,
def configure_logging(debug=False, syslog=False, silenced_loggers=None,
level_overrides=None):
if silenced_loggers is None:
silenced_loggers = []
if level_overrides is None:
level_overrides = {}
level = 'DEBUG' if debug else 'INFO'
renderers = [
dev.ConsoleRenderer(),
] if debug else [
logstash_processor,
processors.JSONRenderer(separators=(',', ':')),
add_syslog_program(syslog),
]
structlog_processors = [
stdlib.filter_by_level,
stdlib.add_logger_name,
stdlib.add_log_level,
fix_logger_name,
format_request,
ensure_event,
stdlib.PositionalArgumentsFormatter(),
processors.TimeStamper(fmt="ISO", key='@timestamp'),
processors.StackInfoRenderer(),
processors.format_exc_info,
},
},
"formatters": {
"plain": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": _chain(structlog.dev.ConsoleRenderer(colors=False), redact),
"foreign_pre_chain": processors,
},
"json": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": _chain(structlog.processors.JSONRenderer(), redact),
"foreign_pre_chain": processors,
},
"colorized": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": _chain(structlog.dev.ConsoleRenderer(colors=True), redact),
"foreign_pre_chain": processors,
},
"debug": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": _chain(structlog.processors.JSONRenderer(), redact),
"foreign_pre_chain": processors,
},
},
"handlers": handlers,
"loggers": {"": {"handlers": handlers.keys(), "propagate": True}},
}
)
structlog.configure(
processors=processors + [structlog.stdlib.ProcessorFormatter.wrap_for_formatter],
wrapper_class=structlog.stdlib.BoundLogger,
logger_factory=structlog.stdlib.LoggerFactory(),
logging.getLogger("urllib3").setLevel("INFO")
shared_processors = [
format_to_hex,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S.%f"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
]
if log_json:
processors = shared_processors + [structlog.processors.JSONRenderer()]
else:
processors = shared_processors + [structlog.dev.ConsoleRenderer()]
structlog.configure(
processors=processors,
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)