Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import logging.config
import logging
import structlog
logging.basicConfig(
level=logging.DEBUG,
format='%(levelname)s[%(threadName)s] %(message)s',
)
logging.getLogger('urllib3').setLevel(logging.CRITICAL)
logging.getLogger('botocore').setLevel(logging.CRITICAL)
structlog.configure_once(
context_class=structlog.threadlocal.wrap_dict(dict),
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
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.stdlib.render_to_log_kwargs]
)
level_map = {
'CRITICAL': 50,
'ERROR': 40,
'WARNING': 30,
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",utc=True),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
# structlog.dev.ConsoleRenderer(colors=True)
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,
)
# pylint: enable=no-member
LOG_DATETIME_FORMAT = r'%Y-%m-%dT%H:%M:%S.%s%Z'
os.environ['TZ'] = 'UTC'
time.tzset()
# JsonFormatter.converter = time.gmtime
SUPPORTED_LOG_MESSAGE_KEYS = (
'levelname',
'asctime',
# 'created',
# 'filename',
# 'levelno',
# 'module',
import ujson
import structlog
from pythonjsonlogger.jsonlogger import JsonFormatter
# pylint: disable=c-extension-no-member
import rapidjson
from jussi.typedefs import WebApp
# pylint: disable=no-member
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",utc=True),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
# structlog.dev.ConsoleRenderer(colors=True)
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,
)
# pylint: enable=no-member
LOG_DATETIME_FORMAT = r'%Y-%m-%dT%H:%M:%S.%s%Z'
os.environ['TZ'] = 'UTC'
# Add TRACE log level (lower than DEBUG:10)
TRACE_LOGLVL = 5
logging.addLevelName(TRACE_LOGLVL, "TRACE")
def trace_loglevel(self, message, *args, **kws):
if self.isEnabledFor(TRACE_LOGLVL):
self._log(TRACE_LOGLVL, message, args, **kws)
logging.Logger.trace = trace_loglevel
processors = [
add_exc_info_flag_for_exception,
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
add_instance_id,
StructuredLogRenderer(),
]
structlog.configure(logger_factory=structlog.stdlib.LoggerFactory(),
context_class=PlainRenderedOrderedDict,
wrapper_class=BoundLogger,
processors=processors)
# Mark first line of log
log = structlog.get_logger()
log.info("first-line")
return log
def __init__(self):
WrappedDictClass = structlog.threadlocal.wrap_dict(dict)
structlog.configure(
processors=[
self.add_log_level,
structlog.processors.TimeStamper(
fmt="iso", utc=False, key="time"), self.event2message,
structlog.processors.JSONRenderer()
],
context_class=WrappedDictClass(),
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True, )
self.log = structlog.get_logger("event")
def configure():
conf(
context_class=threadlocal.wrap_dict(dict),
logger_factory=stdlib.LoggerFactory(),
wrapper_class=stdlib.BoundLogger,
processors=[
stdlib.PositionalArgumentsFormatter(),
processors.TimeStamper(fmt="iso"),
processors.StackInfoRenderer(),
processors.format_exc_info,
processors.UnicodeDecoder(),
stdlib.render_to_log_kwargs,
]
import 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,
structlog.processors.JSONRenderer(),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
logger = logging.getLogger("flowmachine").getChild(__name__)
# Logger for all queries run or accessed
query_run_log = logging.getLogger("flowmachine-server")
query_run_log.setLevel(logging.INFO)
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
query_run_log.addHandler(ch)
log_root = os.getenv("LOG_DIRECTORY", "/var/log/flowmachine/")
if not os.path.exists(log_root):
logger.info(f"Creating log_root directory because it does not exist: {log_root}")
os.makedirs(log_root)
fh = TimedRotatingFileHandler(os.path.join(log_root, "query-runs.log"), when="midnight")
# ...
return "logged in!"
if __name__ == "__main__":
logging.basicConfig(
format="%(message)s", stream=sys.stdout, level=logging.INFO
)
structlog.configure(
processors=[
structlog.processors.KeyValueRenderer(
key_order=["event", "request_id"]
)
],
context_class=structlog.threadlocal.wrap_dict(dict),
logger_factory=structlog.stdlib.LoggerFactory(),
)
app.run()
"formatter": "key_value",
},
},
"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,
)