# SPDX-License-Identifier: MIT OR Apache-2.0 # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the MIT License. See the LICENSE file in the root of this # repository for complete details. """ Make sure our configuration examples actually pass the type checker. """ from __future__ import annotations import logging import logging.config from typing import Any, Callable import structlog from structlog.processors import CallsiteParameter from structlog.typing import FilteringBoundLogger bl = structlog.get_logger() bl.msg("hello", whom="world", x=42, y={}) bls: structlog.stdlib.BoundLogger = structlog.get_logger() bls.info("hello", whom="world", x=42, y={}) def bytes_dumps( __obj: Any, /, default: Callable[[Any], Any] | None = None, option: int | None = None, ) -> bytes: """ Test with orjson's signature taken from . """ return b"{}" structlog.configure( processors=[structlog.processors.JSONRenderer(serializer=bytes_dumps)] ) structlog.configure( processors=[ structlog.stdlib.render_to_log_args_and_kwargs, structlog.stdlib.filter_by_level, 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, ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) structlog.configure( processors=[ structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), ) formatter = structlog.stdlib.ProcessorFormatter( processor=structlog.dev.ConsoleRenderer(), ) formatter = structlog.stdlib.ProcessorFormatter( processors=[ structlog.processors.CallsiteParameterAdder(), structlog.processors.CallsiteParameterAdder( set(CallsiteParameter), ["threading"] ), structlog.processors.CallsiteParameterAdder( set(CallsiteParameter), additional_ignores=["threading"] ), structlog.processors.CallsiteParameterAdder( parameters=set(CallsiteParameter), additional_ignores=["threading"] ), structlog.processors.CallsiteParameterAdder( [ CallsiteParameter.FILENAME, CallsiteParameter.FUNC_NAME, CallsiteParameter.LINENO, ] ), structlog.processors.CallsiteParameterAdder( parameters=[ CallsiteParameter.FILENAME, CallsiteParameter.FUNC_NAME, CallsiteParameter.LINENO, ] ), structlog.processors.CallsiteParameterAdder( parameters=[ CallsiteParameter.FILENAME, CallsiteParameter.FUNC_NAME, CallsiteParameter.LINENO, ] ), ], ) handler = logging.StreamHandler() handler.setFormatter(formatter) root_logger = logging.getLogger() root_logger.addHandler(handler) root_logger.setLevel(logging.INFO) timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S") shared_processors: list[structlog.typing.Processor] = [ structlog.stdlib.add_log_level, timestamper, ] structlog.configure( processors=[ *shared_processors, structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), cache_logger_on_first_use=True, ) formatter = structlog.stdlib.ProcessorFormatter( processor=structlog.dev.ConsoleRenderer(), foreign_pre_chain=[ structlog.stdlib.ExtraAdder(), structlog.stdlib.ExtraAdder(allow=None), structlog.stdlib.ExtraAdder(None), structlog.stdlib.ExtraAdder(allow=["k1", "k2"]), structlog.stdlib.ExtraAdder({"k1", "k2"}), *shared_processors, ], ) timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S") pre_chain = [ # Add the log level and a timestamp to the event_dict if the log entry # is not from structlog. structlog.stdlib.add_log_level, timestamper, ] 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, }, "colored": { "()": structlog.stdlib.ProcessorFormatter, "processor": structlog.dev.ConsoleRenderer(colors=True), "foreign_pre_chain": pre_chain, }, }, "handlers": { "default": { "level": "DEBUG", "class": "logging.StreamHandler", "formatter": "colored", }, "file": { "level": "DEBUG", "class": "logging.handlers.WatchedFileHandler", "filename": "test.log", "formatter": "plain", }, }, "loggers": { "": { "handlers": ["default", "file"], "level": "DEBUG", "propagate": True, }, }, } ) structlog.configure( processors=[ structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), timestamper, structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) 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.JSONRenderer(), ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) 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.LogfmtRenderer(), ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) 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.JSONRenderer(), ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.AsyncBoundLogger, cache_logger_on_first_use=True, ) # Regression test for # https://github.com/wemake-services/wemake-django-template/ 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, ) with structlog.threadlocal.bound_threadlocal(x=42): pass def typecheck_filtering_return() -> None: fblogger: FilteringBoundLogger = structlog.get_logger(__name__) fblog = fblogger.bind(key1="value1", key2="value2", key3="value3") fblog.info("values bound") fblog = fblog.unbind("key1") fblog.debug("value unbound") fblog = fblog.try_unbind("bad_key") fblog.warn("no value unbound because key not defined") fblog = fblog.new(new="value") fblog.info("this is a whole new logger") fblog.log(logging.CRITICAL, "this is synchronously CRITICAL") async def typecheck_filtering_return_async() -> None: fblogger: FilteringBoundLogger = structlog.get_logger(__name__) await fblogger.adebug("async debug") await fblogger.ainfo("async info") await fblogger.awarning("async warning") await fblogger.awarn("async warn") await fblogger.aerror("async error") await fblogger.afatal("fatal error") await fblogger.aexception("async exception") await fblogger.acritical("async critical") await fblogger.amsg("async msg") await fblogger.alog(logging.CRITICAL, "async log") async def typecheck_stdlib_async() -> None: logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) await logger.adebug("async debug") await logger.ainfo("async info") await logger.awarning("async warning") await logger.aerror("async error") await logger.afatal("fatal error") await logger.aexception("async exception") await logger.acritical("async critical") await logger.alog(logging.CRITICAL, "async log") def typecheck_bound_logger_return() -> None: blogger: structlog.BoundLogger = structlog.get_logger(__name__) blog = blogger.bind(key1="value1", key2="value2", key3="value3") blog = blog.unbind("key1") blog = blog.try_unbind("bad_key") blog = blog.new(new="value") # Structured tracebacks and ExceptionRenderer with ExceptionDictTransformer struct_tb: structlog.tracebacks.Trace = structlog.tracebacks.extract( ValueError, ValueError("onoes"), None ) try: raise ValueError("onoes") except ValueError as e: struct_tb = structlog.tracebacks.extract(type(e), e, e.__traceback__) structlog.configure( processors=[ structlog.processors.ExceptionRenderer( structlog.tracebacks.ExceptionDictTransformer() ), structlog.processors.JSONRenderer(), ] ) fbl: FilteringBoundLogger = structlog.get_logger() fbl.info("Hello %s! The answer is %d.", "World", 42, x=1) # Introspection level: int = fbl.get_effective_level() is_active: bool = fbl.is_enabled_for(logging.INFO) is_active = fbl.is_enabled_for(20)