{"id":1008,"date":"2026-05-31T16:58:06","date_gmt":"2026-05-31T08:58:06","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=1008"},"modified":"2026-05-31T16:58:06","modified_gmt":"2026-05-31T08:58:06","slug":"a-coding-implementation-on-loguru-for-designing-robust-structured-concurrent-and-production-ready-python-logging-pipelines","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=1008","title":{"rendered":"A Coding Implementation on Loguru for Designing Robust, Structured, Concurrent, and Production-Ready Python Logging Pipelines"},"content":{"rendered":"

In this tutorial, we implement a practical use case with Loguru<\/strong><\/a>, a powerful, flexible, and production-ready logging library for Python. We start by building a clean, idempotent logging setup that can be safely rerun without duplicating handlers or producing messy output. From there, we move step by step through structured logging, contextual logging, custom log levels, global patching, callable formatters, and in-memory sinks. We also handle real-world logging needs such as rich exception traces, JSON log files, custom rotation, compression, retention, async logging, threaded execution, multiprocessing-safe logging, and standard logging module interception. By keeping everything in a Colab-ready workflow, we make it easy to test, inspect, and understand how Loguru can support debugging, monitoring, and observability in serious Python applications.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip install -q loguru nest_asyncio\nimport os, sys, time, json, glob, gzip, shutil, asyncio, logging, itertools, multiprocessing\nfrom collections import deque\nfrom concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor\nfrom loguru import logger\ntry:\n   import nest_asyncio; nest_asyncio.apply()\nexcept Exception as e:\n   print(\"nest_asyncio not applied:\", e)\nWORKDIR = \"\/content\/loguru_demo\" if os.path.isdir(\"\/content\") else \"\/tmp\/loguru_demo\"\nos.makedirs(WORKDIR, exist_ok=True); os.chdir(WORKDIR)\nfor f in glob.glob(\"*\"):\n   try: os.remove(f)\n   except OSError: pass\nprint(f\"Working directory: {WORKDIR}n\")\nRESULTS = []\ndef check(name, condition, detail=\"\"):\n   ok = bool(condition); RESULTS.append((name, ok))\n   print(f\"   [{'PASS' if ok else 'FAIL'}] {name}\" + (f\" \u2014 {detail}\" if detail else \"\"))\ndef banner(t): print(f\"n{'='*64}n  {t}n{'='*64}\")\n_seq = itertools.count(1)\ndef global_patcher(record):\n   record[\"extra\"].setdefault(\"env\", \"colab\")\n   record[\"extra\"][\"seq\"] = next(_seq)\n_NOISE = {\"env\", \"seq\", \"app\"}\ndef console_formatter(record):\n   fmt = (\"<green>{time:HH:mm:ss.SSS}<\/green> | <level>{level: <8}<\/level> | \"\n          \"<cyan>{name}:{function}:{line}<\/cyan> - <level>{message}<\/level>\")\n   if any(k not in _NOISE for k in record[\"extra\"]):\n       fmt += \" | <yellow>{extra}<\/yellow>\"\n   return fmt + \"n{exception}\"<\/code><\/pre>\n<\/div>\n<\/div>\n
We install Loguru and supporting dependencies, import all required libraries, and prepare a clean working directory for the tutorial. We also create a small verification helper to test each feature as the tutorial runs. We then define a global patcher and console formatter so that every log record carries useful metadata and appears in a readable format.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class MemorySink:\n   def __init__(self, capacity=2000): self.buffer = deque(maxlen=capacity)\n   def write(self, message): self.buffer.append(message.record)\n   def flush(self): pass\n   def has_level(self, name): return any(r[\"level\"].name == name for r in self.buffer)\n   def find(self, pred):      return [r for r in self.buffer if pred(r)]\nMAX_BYTES = 1500\ndef size_rotation(message, file):\n   return file.tell() + len(message) > MAX_BYTES\ndef gzip_compression(filepath):\n   with open(filepath, \"rb\") as fi, gzip.open(filepath + \".gz\", \"wb\") as fo:\n       shutil.copyfileobj(fi, fo)\n   os.remove(filepath)\ndef keep_latest_retention(files):\n   for old in sorted(files, key=os.path.getmtime, reverse=True)[3:]:\n       try: os.remove(old)\n       except OSError: pass\nclass InterceptHandler(logging.Handler):\n   def emit(self, record):\n       try: level = logger.level(record.levelname).name\n       except ValueError: level = record.levelno\n       frame, depth = logging.currentframe(), 2\n       while frame and frame.f_code.co_filename == logging.__file__:\n           frame, depth = frame.f_back, depth + 1\n       (logger.opt(depth=depth, exception=record.exc_info)\n              .bind(stdlib_logger=record.name)\n              .log(level, record.getMessage()))\ndef mp_worker(n):\n   logger.bind(child=os.getpid()).info(\"hello from child item {}\", n)\n   return os.getpid()<\/code><\/pre>\n<\/div>\n<\/div>\n
We create reusable logging components that make the tutorial more practical and production-like. We define an in-memory sink, custom file rotation, compression, and retention functions to control how logs are stored. We also built a standard logging interceptor and a multiprocessing worker to connect Loguru to external libraries and child processes.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
banner(\"1) logger.configure(): handlers + custom level + extra + patcher\")\nmem = MemorySink()\nlogger.configure(\n   handlers=[\n       {\"sink\": sys.stderr, \"format\": console_formatter, \"level\": \"DEBUG\",\n        \"colorize\": True, \"backtrace\": True, \"diagnose\": True},\n       {\"sink\": mem, \"level\": \"DEBUG\", \"format\": \"{message}\"},\n       {\"sink\": \"structured.jsonl\", \"serialize\": True, \"level\": \"DEBUG\",\n        \"enqueue\": True},\n       {\"sink\": \"errors.log\", \"level\": \"ERROR\", \"enqueue\": True,\n        \"backtrace\": True, \"diagnose\": False,\n        \"format\": \"{time:YYYY-MM-DD HH:mm:ss} | {level} | \"\n                  \"{name}:{function}:{line} | {message}\"},\n   ],\n   levels=[{\"name\": \"NOTICE\", \"no\": 22, \"color\": \"<blue><bold>\", \"icon\": \"\"\ud83d\udce2\"\"}],\n   extra={\"app\": \"loguru-advanced\"},\n   patcher=global_patcher,\n)\nlogger.debug(\"debug\"); logger.info(\"info\"); logger.success(\"SUCCESS level ships built-in\")\nlogger.warning(\"warning\"); logger.log(\"NOTICE\", \"custom level between INFO and SUCCESS\")\nbanner(\"2) bind() \/ contextualize() \/ patch()\")\nlogger.bind(user_id=42, request_id=\"abc-123\").info(\"bound context\")\nwith logger.contextualize(task=\"batch-job\", run=7):\n   logger.info(\"inside contextualized block\")\nlogger.patch(lambda r: r[\"extra\"].update(epoch=round(time.time()))).info(\"per-call patched record\")\nbanner(\"3) @logger.catch + context-manager form\")\ndef inner(d):  return d[\"a\"] \/ d[\"b\"]\ndef outer(d):  return inner(d)\n@logger.catch(reraise=False)\ndef compute(d): return outer(d)\ncompute({\"a\": 1, \"b\": 0})\nwith logger.catch(message=\"handled inside a with-block\"):\n   raise ValueError(\"boom in block\")\nbanner(\"4) opt(lazy=True), inline colors, record access\")\nlogger.opt(lazy=True).debug(\"lazy sum = {}\", lambda: sum(i*i for i in range(1_000_000)))\nlogger.opt(colors=True).info(\"inline <red>colors<\/red> <green>work<\/green>\")\nlogger.opt(record=True).info(\"emitted from source line {record[line]}\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We configure Loguru with multiple handlers, including console output, memory capture, JSON logging, and error logging. We then demonstrate structured logging with bound context, contextual blocks, patched records, and a custom log level. We also explore exception handling and useful opt() features such as lazy evaluation, inline colors, and record access.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
banner(\"5) custom rotation\/compression\/retention (forces real rotation)\")\nev_id = logger.add(\"events_{time:HHmmss_SSS}.log\",\n                  rotation=size_rotation, compression=gzip_compression,\n                  retention=keep_latest_retention, enqueue=True, level=\"DEBUG\",\n                  format=\"{time:HH:mm:ss.SSS} | {level: <8} | {message}\")\nfor i in range(80):\n   logger.bind(idx=i).debug(\"rotating event line number {}\", i)\nlogger.complete(); logger.remove(ev_id)\nprint(f\"   archives created: {sorted(glob.glob('events_*.gz'))}\")\nbanner(\"6a) ThreadPoolExecutor with per-thread contextualize()\")\nthread_caps = []\ntid = logger.add(thread_caps.append, level=\"DEBUG\", format=\"{message}\",\n                filter=lambda r: \"worker_id\" in r[\"extra\"])\ndef worker(n):\n   with logger.contextualize(worker_id=n):\n       logger.info(\"thread work item {}\", n)\n   return n * n\nwith ThreadPoolExecutor(max_workers=8) as ex:\n   sq = list(ex.map(worker, range(8)))\nlogger.complete(); logger.remove(tid)\nworker_ids = {m.record[\"extra\"][\"worker_id\"] for m in thread_caps}\nbanner(\"6b) async coroutine sink + await logger.complete()\")\nasync def run_async_demo():\n   sunk = []\n   async def async_sink(message):\n       await asyncio.sleep(0); sunk.append(message.record[\"message\"])\n   sid = logger.add(async_sink, level=\"DEBUG\", catch=True)\n   async def task(n):\n       with logger.contextualize(coro=n):\n           logger.info(\"async task {} start\", n)\n           await asyncio.sleep(0.01)\n           logger.success(\"async task {} done\", n)\n   await asyncio.gather(*(task(i) for i in range(5)))\n   await logger.complete()\n   logger.remove(sid)\n   return sunk\ntry:\n   async_msgs = asyncio.run(run_async_demo())\nexcept RuntimeError:\n   async_msgs = asyncio.get_event_loop().run_until_complete(run_async_demo())\nprint(f\"   async sink received {len(async_msgs)} messages\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We demonstrate custom file management by automatically rotating, compressing, and retaining log files. We then test thread-safe logging by running multiple workers, each with its own contextual metadata. We also add an asynchronous coroutine sink to see how Loguru handles async tasks and correctly drains pending logs.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
banner(\"7) intercept stdlib `logging` and filter a chatty library\")\nlogging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)\nlib_caps = []\ndef lib_filter(record):\n   if record[\"extra\"].get(\"stdlib_logger\") == \"chatty\":\n       return record[\"level\"].no >= logger.level(\"WARNING\").no\n   return True\nlid = logger.add(lib_caps.append, level=\"DEBUG\", format=\"{message}\",\n                filter=lambda r: (\"stdlib_logger\" in r[\"extra\"]) and lib_filter(r))\nlogging.getLogger(\"chatty\").info(\"noisy info (should be filtered out)\")\nlogging.getLogger(\"chatty\").warning(\"noisy warning (kept)\")\nlogging.getLogger(\"important\").debug(\"important debug (kept)\")\nlogger.complete(); logger.remove(lid)\nbanner(\"8) SELF-TESTS\")\nlogger.complete(); time.sleep(0.2)\ntry:\n   rec = json.loads(open(\"structured.jsonl\").read().splitlines()[-1])\n   check(\"JSON sink serializes records\", {\"text\", \"record\"} <= set(rec))\nexcept Exception as e:\n   check(\"JSON sink serializes records\", False, str(e))\ntry:\n   err_txt = open(\"errors.log\").read()\n   check(\"errors.log captured ZeroDivisionError\", \"ZeroDivisionError\" in err_txt)\nexcept Exception as e:\n   check(\"errors.log captured ZeroDivisionError\", False, str(e))\ncheck(\"custom rotation produced .gz archives\", len(glob.glob(\"events_*.gz\")) >= 1,\n     f\"{len(glob.glob('events_*.gz'))} archive(s)\")\ncheck(\"custom NOTICE level recorded\", mem.has_level(\"NOTICE\"))\ncheck(\"SUCCESS level recorded\",        mem.has_level(\"SUCCESS\"))\ncheck(\"bound user_id=42 present\",      bool(mem.find(lambda r: r[\"extra\"].get(\"user_id\") == 42)))\ncheck(\"contextualize task present\",    bool(mem.find(lambda r: r[\"extra\"].get(\"task\") == \"batch-job\")))\ncheck(\"global patcher stamped env\",    bool(mem.find(lambda r: r[\"extra\"].get(\"env\") == \"colab\")))\ncheck(\"exception captured in a record\",bool(mem.find(lambda r: r[\"exception\"] is not None)))\ncheck(\"threads logged all 8 workers\",  worker_ids == set(range(8)), str(sorted(worker_ids)))\ncheck(\"async sink got 10 messages\",    len(async_msgs) == 10, f\"{len(async_msgs)} msgs\")\nkept = {m.record[\"extra\"][\"stdlib_logger\"] + \":\" + m.record[\"level\"].name for m in lib_caps}\ncheck(\"library INFO filtered, rest kept\",\n     (\"chatty:INFO\" not in kept) and (\"chatty:WARNING\" in kept) and (\"important:DEBUG\" in kept),\n     str(sorted(kept)))<\/code><\/pre>\n<\/div>\n<\/div>\n
We intercept Python\u2019s built-in logging module and route standard library logs into Loguru. We apply source-aware filtering so that noisy logs from one library can be suppressed while important messages are still kept. We then run self-tests to verify JSON logging, error capture, archive creation, context propagation, exception records, threaded logs, async logs, and filtering behavior.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
banner(\"9) throughput: enqueue=False vs enqueue=True\")\ndef bench(enqueue, n=15000):\n   logger.remove()\n   sid = logger.add(lambda m: None, level=\"DEBUG\", format=\"{message}\", enqueue=enqueue)\n   t0 = time.perf_counter()\n   for i in range(n): logger.bind(i=i).debug(\"benchmark {}\", i)\n   logger.complete(); dt = time.perf_counter() - t0\n   logger.remove(sid)\n   return n \/ dt\ntry:\n   sync_tput = bench(False); async_tput = bench(True)\n   print(f\"   direct  : {sync_tput:,.0f} msg\/s\")\n   print(f\"   enqueue : {async_tput:,.0f} msg\/s (non-blocking, process\/thread-safe)\")\n   check(\"benchmark completed\", sync_tput > 0 and async_tput > 0)\nexcept Exception as e:\n   check(\"benchmark completed\", False, str(e))\nbanner(\"10) multiprocessing with enqueue=True (fork)\")\ntry:\n   logger.remove()\n   mp_id = logger.add(\"mp.log\", enqueue=True, level=\"DEBUG\",\n                      format=\"{extra[child]} | {message}\")\n   ctx = multiprocessing.get_context(\"fork\")\n   with ProcessPoolExecutor(max_workers=4, mp_context=ctx) as ex:\n       pids = list(ex.map(mp_worker, range(4)))\n   logger.complete(); logger.remove(mp_id); time.sleep(0.1)\n   lines = open(\"mp.log\").read().splitlines()\n   check(\"multiprocessing logged from children\", len(lines) >= 4,\n         f\"{len(lines)} lines from {len(set(pids))} PIDs\")\nexcept Exception as e:\n   check(\"multiprocessing logged from children\", False, f\"unsupported here: {e}\")\nlogger.remove()\nlogger.add(sys.stderr, level=\"INFO\", colorize=True,\n          format=\"<green>{time:HH:mm:ss}<\/green> | <level>{level}<\/level> | {message}\")\nbanner(\"RESULTS\")\npassed = sum(ok for _, ok in RESULTS)\nfor name, ok in RESULTS:\n   print(f\"   {'\"\u2705\"' if ok else '\"\u274c\"'} {name}\")\nprint(f\"n   {passed}\/{len(RESULTS)} checks passed\")\nprint(f\"   files: {sorted(glob.glob('*'))}\")\n(logger.success if passed == len(RESULTS) else logger.warning)(\n   \"\"\u2705\" Loguru tutorial complete!\" if passed == len(RESULTS)\n   else f\"\"\u26a0\" Completed with {len(RESULTS)-passed} failed check(s)\"\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We benchmark Loguru throughput by comparing direct logging with enqueue-based logging. We then test multiprocessing-safe logging by writing messages from child processes into a shared log file. Also, we clean up the logger, print all test results, list the generated files, and show whether the tutorial completes successfully.<\/p>\n
In conclusion, we built a complete and robust logging system using Loguru that goes far beyond basic print-style debugging. We learned how to configure multiple sinks, capture structured JSON records, add contextual metadata, preserve useful exception information, manage rotating log files, filter noisy third-party libraries, and handle concurrent workloads across threads, async tasks, and processes. We also included self-verification checks and a small benchmark to confirm that the logging pipeline works correctly and to assess its performance behavior.<\/p>\n
\n
\n<\/p>
\n<\/p>
Check out\u00a0the\u00a0Full Codes here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0150k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\n
The post A Coding Implementation on Loguru for Designing Robust, Structured, Concurrent, and Production-Ready Python Logging Pipelines<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we implement…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-1008","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/1008","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1008"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/1008\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1008"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1008"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1008"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}