Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def configure_for_pf():
"""
Configure structlog to use ProcessorFormatter.
Reset both structlog and logging setting after the test.
"""
configure(
processors=[add_log_level, ProcessorFormatter.wrap_for_formatter],
logger_factory=LoggerFactory(),
wrapper_class=BoundLogger,
)
yield
logging.basicConfig()
reset_defaults()
def main(index, project, dry_run, paths):
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M.%S"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
# Coloured output if to terminal.
CleanConsoleRenderer() if sys.stdout.isatty() else structlog.processors.JSONRenderer(),
],
context_class=dict,
cache_logger_on_first_use=True,
)
# TODO: @ui.executor_cli_options
archive_all(
index,
project,
[Path(path).absolute() for path in paths],
dry_run=dry_run,
import sys
import logging
import structlog
from .lib.helpers import env_is_true
from .lib.json import json_def, json_dumps
from pythonjsonlogger import jsonlogger
logging.basicConfig(level=logging.DEBUG, format="%(message)s")
logHandler = logging.StreamHandler(sys.stdout)
logHandler.setFormatter(jsonlogger.JsonFormatter())
processors = [
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
# structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
# structlog.processors.UnicodeDecoder(),
# structlog.stdlib.render_to_log_kwargs,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M.%S", utc=False),
]
if env_is_true('JSON_LOGS'):
processors.append(structlog.processors.JSONRenderer(serializer=json_dumps))
pass
else:
processors.append(structlog.dev.ConsoleRenderer())
},
'error': {
'class': 'logging.StreamHandler',
'stream': sys.stderr,
},
},
}
logging.config.dictConfig(logging_config)
structlog.configure(
processors=[
contextvars.merge_contextvars,
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(),
structlog.processors.KeyValueRenderer(
key_order=['timestamp', 'logger', 'level', 'event']
),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
},
},
"loggers": {
"django_structlog": {
"handlers": ["console", "flat_line_file", "json_file"],
"level": "INFO",
},
}
}
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.processors.ExceptionPrettyPrinter(),
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
context_class=structlog.threadlocal.wrap_dict(dict),
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
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,
] + renderers
configure(
processors=structlog_processors,
context_class=dict,
logger_factory=stdlib.LoggerFactory(),
wrapper_class=stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
import logging
import sys
import structlog
LOGGER_NAME = "rigor"
DEFAULT_PROCESSORS = [
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M.%S"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
]
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))
if isinstance(profiles, dict):
profiles_map = {name: Profile(name, **profile_kwargs)
for name, profile_kwargs in profiles.items()}
else:
profiles_map = {profile.name: profile for profile in profiles or []}
PROFILES.update(profiles_map)
DOCKER_URL = docker_url
# structlog.is_configured() was added in 18.1
if not structlog._config._CONFIG.is_configured:
structlog.configure(
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.KeyValueRenderer(key_order=['event']),
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
def configure_logger(log_name, log_dir, log_level):
eventrenamer = EventRenamer("message")
timestamper = structlog.processors.TimeStamper(fmt="iso", utc=False, key="@timestamp")
shared_processors = [
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
add_thread_info,
timestamper,
eventrenamer
]
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
add_thread_info,
structlog.stdlib.PositionalArgumentsFormatter(),
timestamper,
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.stdlib.render_to_log_kwargs,
#eventrenamer
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
formatter = structlog.stdlib.ProcessorFormatter(
def configure_logging(logging_levels, plain=False, stream=sys.stderr):
renderer = (
PlainRenderer() if plain else
structlog.processors.JSONRenderer())
attr_processors = [
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt='iso')
]
structlog.configure_once(
processors=(
[structlog.stdlib.filter_by_level] +
attr_processors +
[
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.stdlib.ProcessorFormatter.wrap_for_formatter
]
),
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),