Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
- Optionally cache the logger on first use
:return: structured logger
"""
def add_instance_id(_, __, event_dict):
event_dict['instance_id'] = instance_id
return event_dict
# Configure standard logging
logging.config.dictConfig(log_config)
logging.root.level -= 10 * verbosity_adjust
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
add_instance_id,
structlog.processors.UnicodeEncoder(),
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=cache_on_use,
)
# Mark first line of log
log = structlog.get_logger()
log.info("first-log-line, logging level %d" % logging.root.level)
return log
# Log to papertrail
if channel and PAPERTRAIL_HOST and PAPERTRAIL_PORT:
setup_papertrail(project_name, channel, PAPERTRAIL_HOST, PAPERTRAIL_PORT)
# Log to senty
if channel and SENTRY_DSN:
setup_sentry(project_name, channel, SENTRY_DSN)
def logbook_factory(*args, **kwargs):
# Logger given to structlog
logbook.compat.redirect_logging()
return logbook.Logger(level=level, *args, **kwargs)
# Setup structlog over logbook, with args list at the end
processors = [
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
UnstructuredRenderer(),
]
structlog.configure(
context_class=structlog.threadlocal.wrap_dict(dict),
processors=processors,
logger_factory=logbook_factory,
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
def configure_logging():
"""Configure logging."""
logging.basicConfig(stream=sys.stderr, format="%(message)s", level=logging.CRITICAL)
logging.getLogger("greynoise").setLevel(logging.WARNING)
structlog.configure(
processors=[
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,
structlog.dev.ConsoleRenderer(),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.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))
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True
SECRET_KEY = "dont-tell-eve"
AWS_ACCESS_KEY_ID = ""
AWS_SECRET_ACCESS_KEY = ""
AWS_S3_BUCKET_NAME = "pdfs.contratospr.com"
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.UnicodeDecoder(),
structlog.processors.KeyValueRenderer(),
],
context_class=structlog.threadlocal.wrap_dict(dict),
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
:param vcore_id: The assigned vcore id
:return: structured logger
"""
def add_instance_id(_, __, event_dict):
event_dict['instance_id'] = instance_id
return event_dict
def add_vcore_id(_, __, event_dict):
event_dict['vcore_id'] = vcore_id
return event_dict
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
add_instance_id,
add_vcore_id,
structlog.processors.UnicodeEncoder(),
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=cache_on_use,
)
# Mark first line of log
log = structlog.get_logger()
log.info("updated-logger")
def setup_structlog():
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,
# Uncomment ONE Renderer:
#structlog.processors.KeyValueRenderer(),
#structlog.processors.JSONRenderer(),
structlog.dev.ConsoleRenderer()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
debug_logger.addHandler(ch)
# Logger for all queries run or accessed (used by flowmachine server)
query_run_log = logging.getLogger("flowmachine").getChild("query_run_log")
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(logging.INFO)
query_run_log.addHandler(ch)
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.JSONRenderer(serializer=rapidjson.dumps),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
FLOWKIT_LOGGERS_HAVE_BEEN_INITIALISED = True
class TQDMLogger:
def msg(self, message):
tqdm.write(message)
log = debug = info = warm = warning = msg
fatal = failure = err = error = critical = exception = msg
running_interactively = sys.stdout.isatty() and os.environ.get('PBS_NCPUS', None) is None
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
add_proc_info,
add_mpi_rank,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.dev.ConsoleRenderer() if running_interactively else structlog.processors.JSONRenderer(),
],
context_class=dict,
logger_factory=tqdm_logger_factory if running_interactively else structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)