Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
rv = cr(
None, None, {"event": "test", "level": "critical", "foo": "bar"}
)
# fmt: off
assert (
"[" + dev.RED + styles.bright +
dev._pad("critical", cr._longest_level) +
styles.reset + "] " +
padded +
styles.kv_key + "foo" + styles.reset + "=" +
styles.kv_value + "bar" + styles.reset
) == rv
# fmt: on
assert not dev._has_colorama or dev._ColorfulStyles is cr._styles
def test_colorama_force_colors(self, styles, padded):
"""
If force_colors is True, use colors even if the destination is non-tty.
"""
cr = dev.ConsoleRenderer(
colors=dev._has_colorama, force_colors=dev._has_colorama
)
rv = cr(
None, None, {"event": "test", "level": "critical", "foo": "bar"}
)
# fmt: off
assert (
"[" + dev.RED + styles.bright +
dev._pad("critical", cr._longest_level) +
styles.reset + "] " +
padded +
styles.kv_key + "foo" + styles.reset + "=" +
styles.kv_value + "bar" + styles.reset
) == rv
# fmt: on
def test_level(self, cr, styles, padded):
"""
Levels are rendered aligned, in square brackets, and color coded.
"""
rv = cr(
None, None, {"event": "test", "level": "critical", "foo": "bar"}
)
# fmt: off
assert (
"[" + dev.RED + styles.bright +
dev._pad("critical", cr._longest_level) +
styles.reset + "] " +
padded +
styles.kv_key + "foo" + styles.reset + "=" +
styles.kv_value + "bar" + styles.reset
) == rv
# fmt: on
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())
structlog.configure(
processors=processors,
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
# cache_logger_on_first_use=True,
)
logger = structlog.get_logger()
x__all__ = ['logger']
DATABASES["default"]["ATOMIC_REQUESTS"] = True
# Your stuff...
# ------------------------------------------------------------------------------
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"json_formatter": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.processors.JSONRenderer(),
},
"colored": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer(colors=True),
},
"key_value": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.processors.KeyValueRenderer(
key_order=["timestamp", "level", "event", "logger"]
),
},
},
"handlers": {
"colored_stream": {"class": "logging.StreamHandler", "formatter": "colored"},
"json_file": {
"class": "logging.handlers.WatchedFileHandler",
"filename": "logs/json.log",
"formatter": "json_formatter",
},
"flat_line_file": {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'aws': {
# No time b.c. CloudWatch logs times
'format': u"[%(levelname)-8s] %(message)s "
u"{%(module)s.%(funcName)s():%(lineno)s %(pathname)s}",
'datefmt': "%Y-%m-%d %H:%M:%S"
},
"colored": {
'format': '{Time: %(asctime)s, '
'Level: [%(levelname)s], '
'function: %(module)s.%(funcName)s():%(lineno)s, '
'message: %(message)s}',
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer(colors=True),
'datefmt': '%Y-%m-%d %H:%M:%S',
}},
'handlers': loggingHandlersConfig,
'root': {
'level': 'INFO',
'propagate': True,
'handlers': loggingHandlers
}
}
dictConfig(loggingConfig)
app = Flask(__name__)
# HTTP security header middleware for Flask
talisman = Talisman(app)
talisman.force_https = False
github_interface = make_github_interface(config)
]
logging.config.dictConfig({
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"plain": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer(colors=False),
"foreign_pre_chain": pre_chain,
"keep_exc_info": True,
"keep_stack_info": True
},
"colored": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer(colors=True),
"foreign_pre_chain": pre_chain,
"keep_exc_info": True,
"keep_stack_info": True
},
},
"handlers": {
"default": {
"level": os.environ.get('LOG_LEVEL','DEBUG'),
"class": "logging.StreamHandler",
"formatter": "colored",
},
},
"loggers": {
"": {
"handlers": ["default"],
"level": os.environ.get('LOG_LEVEL','DEBUG'),
from .sbds_json import dumps
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.processors.JSONRenderer(serializer=dumps)
structlog.dev.ConsoleRenderer()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=os.environ.get('LOG_LEVEL', 'INFO'),
)
def session_to_dict(session):
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,
)
from sbds.storages.s3.cli import s3
from sbds.storages.fs.cli import fs
from sbds.codegen.cli import codegen
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()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
logging.basicConfig(
format="%(message)s",
stream=sys.stderr,
level=os.environ.get('LOG_LEVEL', 'INFO'),
)
@click.group(
short_help='manages storage, retrieval, and querying of the Steem blockchain')