Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_bind_before_add(writer):
logger_bound = logger.bind(a=0)
logger.add(writer, format="{extra[a]} {message}")
logger_bound.debug("A")
assert writer.read() == "0 A\n"
def test_not_caught_exception_sink_write(capsys):
logger.add(NotWritable(), enqueue=True, catch=False, format="{message}")
logger.info("It's fine")
logger.bind(fail=True).info("Bye bye...")
logger.info("It's not fine")
logger.remove()
out, err = capsys.readouterr()
lines = err.strip().splitlines()
assert out == "It's fine\n"
assert lines[0].startswith("Exception")
assert lines[-1] == "RuntimeError: You asked me to fail..."
def test_ansi_dont_color_unrelated(writer):
logger.add(writer, format="{message} {extra[trap]}", colorize=True)
logger.bind(trap="B").opt(ansi=True).debug("A")
assert writer.read() == parse("A") + " B\n"
def test_caught_exception_queue_get(writer, capsys):
logger.add(writer, enqueue=True, catch=True, format="{message}")
logger.info("It's fine")
logger.bind(broken=NotUnpicklable()).info("Bye bye...")
logger.info("It's fine again")
logger.remove()
out, err = capsys.readouterr()
lines = err.strip().splitlines()
assert writer.read() == "It's fine\nIt's fine again\n"
assert out == ""
assert lines[0] == "--- Logging error in Loguru Handler #0 ---"
assert lines[1] == "Record was: None"
assert lines[-2] == "RuntimeError: You shall not de-serialize me!"
assert lines[-1] == "--- End of logging error ---"
def test_override_previous_bound(writer):
logger.add(writer, format="{extra[x]} {message}")
logger.bind(x=1).bind(x=2).debug("3")
assert writer.read() == "2 3\n"
def test_unprintable_record(writer, capsys):
class Unprintable:
def __repr__(self):
raise ValueError("Failed")
logger.add(writer, format="{message} {extra[unprintable]}", catch=True)
logger.bind(unprintable=1).debug("a")
logger.bind(unprintable=Unprintable()).debug("b")
logger.bind(unprintable=2).debug("c")
out, err = capsys.readouterr()
lines = err.strip().splitlines()
assert writer.read() == "a 1\nc 2\n"
assert out == ""
assert lines[0] == "--- Logging error in Loguru Handler #0 ---"
assert lines[1] == "Record was: /!\\ Unprintable record /!\\"
assert lines[-2] == "ValueError: Failed"
assert lines[-1] == "--- End of logging error ---"
def test_configure_after_bind(writer):
logger_a = logger.bind(a="A")
logger_b = logger.bind(b="B")
logger.configure(extra={"a": "default_a", "b": "default_b"})
logger.add(writer, format="{extra[a]} {extra[b]} {message}")
logger.debug("init")
logger_a.debug("aaa")
logger_b.debug("bbb")
assert writer.read() == ("default_a default_b init\n" "A default_b aaa\n" "default_a B bbb\n")
def emit(self, record):
# Get corresponding Loguru level if it exists
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message
frame, depth = logging.currentframe(), 2
while frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
logger.bind(name=record.name).opt(depth=depth, exception=record.exc_info).log(
level, record.getMessage()
)
from loguru import logger
class LogType:
MAIN = 1
PRODUCER = 2
MODEL = 3
MCTS = 4
logger.add("log/main.log", filter=lambda r: r["extra"].get("type") == LogType.MAIN)
logger.add("log/producer.log", filter=lambda r: r["extra"].get("type") == LogType.PRODUCER)
logger.add("log/model.log", filter=lambda r: r["extra"].get("type") == LogType.MODEL)
logger.add("log/mcts.log", filter=lambda r: r["extra"].get("type") == LogType.MCTS)
main_logger = logger.bind(type=LogType.MAIN)
producer_logger = logger.bind(type=LogType.PRODUCER)
model_logger = logger.bind(type=LogType.MODEL)
mcts_logger = logger.bind(type=LogType.MCTS)
logger.disable("mcts")