Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import datetime
from concurrent.futures.thread import ThreadPoolExecutor
from contextlib import contextmanager
from multiprocessing import cpu_count
import sqlalchemy as sqlalchemy
from sqlalchemy.exc import ResourceClosedError
import structlog
import json
structlog.configure(
processors=[
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(serializer=json.dumps),
]
)
logger = structlog.get_logger(__name__)
parser = argparse.ArgumentParser(description="Flowminder Synthetic CDR Generator\n")
parser.add_argument(
"--n-subscribers", type=int, default=4000, help="Number of subscribers to generate."
)
parser.add_argument(
"--n-tacs", type=int, default=4000, help="Number of phone models to generate."
)
parser.add_argument(
"--n-sites", type=int, default=1000, help="Number of sites to generate."
)
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,
)
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,
)
'CHILD_CONTEXT_MANAGERS': [],
# Store traceback in execution history for failed tasks. This can
# increase Redis storage requirements and therefore can be disabled
# if that is a concern.
'STORE_TRACEBACKS': True,
}
if config:
self.config.update(config)
if setup_structlog:
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.stdlib.filter_by_level,
structlog.processors.TimeStamper(fmt='iso', utc=True),
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,
)
self.log = structlog.get_logger(self.config['LOGGER_NAME']).bind()
if setup_structlog:
self.log.setLevel(logging.DEBUG)
logging.basicConfig(format='%(message)s')
self.connection = connection or redis.Redis(decode_responses=True)
def __setup_basic_logging(cls, level):
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,
'INFO': 20,
'DEBUG': 10,
'NOTSET': 0,
}
dict_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
from django.conf import settings
import logging.config
import structlog
from sentry import options
from sentry.logging import LoggingFormat
WrappedDictClass = structlog.threadlocal.wrap_dict(dict)
kwargs = {
"context_class": WrappedDictClass,
"wrapper_class": structlog.stdlib.BoundLogger,
"cache_logger_on_first_use": True,
"processors": [
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.format_exc_info,
structlog.processors.StackInfoRenderer(),
structlog.processors.UnicodeDecoder(),
],
}
fmt_from_env = os.environ.get("SENTRY_LOG_FORMAT")
if fmt_from_env:
settings.SENTRY_OPTIONS["system.logging-format"] = fmt_from_env.lower()
fmt = options.get("system.logging-format")
if fmt == LoggingFormat.HUMAN:
from sentry.logging.handlers import HumanRenderer
kwargs["processors"].extend(
[structlog.processors.ExceptionPrettyPrinter(), HumanRenderer()]
)
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,
]
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 setup_logging(log_level: str, log_json: bool) -> None:
""" Basic structlog setup """
logging.basicConfig(level=log_level, stream=sys.stdout, format="%(message)s")
logging.getLogger("web3").setLevel("INFO")
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,
)