{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create custom log handlers\n",
    "\n",
    "**The question.** You need log records to go somewhere the built-in handlers don't reach — an in-memory list for assertions in a test, a CSV file for later analysis, a Slack webhook, a metrics system. The stock `StreamHandler`, `FileHandler`, and `RotatingFileHandler` don't cover it.\n",
    "\n",
    "The answer: subclass `logging.Handler` and override `emit(record)`. The base class takes care of levels, formatters, and thread-safe dispatch; you just decide what to do with the already-formatted message.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The minimal shape: one class, one overridden emit().\n",
    "# This ListHandler captures messages in-process — useful for tests and for\n",
    "# programmatic inspection of what got logged.\n",
    "import logging\n",
    "\n",
    "\n",
    "class ListHandler(logging.Handler):\n",
    "    '''Captures formatted log messages in a list (great for tests).'''\n",
    "\n",
    "    def __init__(self, level: int = logging.NOTSET) -> None:\n",
    "        super().__init__(level)\n",
    "        self.messages: list[str] = []\n",
    "\n",
    "    def emit(self, record: logging.LogRecord) -> None:\n",
    "        try:\n",
    "            self.messages.append(self.format(record))\n",
    "        except Exception:\n",
    "            self.handleError(record)    # never let handler errors escape\n",
    "\n",
    "\n",
    "# Usage — attach to any logger like a built-in handler.\n",
    "logger = logging.getLogger('demo')\n",
    "logger.setLevel(logging.DEBUG)\n",
    "handler = ListHandler()\n",
    "handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))\n",
    "logger.addHandler(handler)\n",
    "\n",
    "logger.info('first message')\n",
    "logger.warning('second message')\n",
    "logger.error('third message')\n",
    "\n",
    "for msg in handler.messages:\n",
    "    print(' ', msg)\n",
    "\n",
    "logger.removeHandler(handler)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Variant: a CSV handler\n",
    "\n",
    "Writes each record as a row: timestamp, level, logger name, message. Useful for post-hoc analysis in pandas or Excel without parsing a free-form log format.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import logging\n",
    "from datetime import datetime, timezone\n",
    "from pathlib import Path\n",
    "\n",
    "\n",
    "class CSVHandler(logging.Handler):\n",
    "    '''Writes records as CSV rows — one row per emit.'''\n",
    "\n",
    "    def __init__(self, path: str | Path, level: int = logging.NOTSET) -> None:\n",
    "        super().__init__(level)\n",
    "        self.path = Path(path)\n",
    "        with self.path.open('w', newline='', encoding='utf-8') as f:\n",
    "            csv.writer(f).writerow(['timestamp', 'level', 'logger', 'message'])\n",
    "\n",
    "    def emit(self, record: logging.LogRecord) -> None:\n",
    "        try:\n",
    "            ts = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat()\n",
    "            with self.path.open('a', newline='', encoding='utf-8') as f:\n",
    "                csv.writer(f).writerow([ts, record.levelname, record.name, record.getMessage()])\n",
    "        except Exception:\n",
    "            self.handleError(record)\n",
    "\n",
    "\n",
    "import tempfile\n",
    "csv_path = Path(tempfile.mkdtemp()) / 'audit.csv'\n",
    "logger = logging.getLogger('csv-demo')\n",
    "logger.setLevel(logging.DEBUG)\n",
    "logger.addHandler(CSVHandler(csv_path))\n",
    "\n",
    "logger.info('service started')\n",
    "logger.warning('cache miss for key %r', 'user:42')\n",
    "\n",
    "print(csv_path.read_text())\n",
    "logger.handlers.clear()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Variant: filters for content-based routing\n",
    "\n",
    "Attach a `logging.Filter` to a handler (or a logger) to accept/reject records based on anything you can compute from the record. Here, only records whose message contains 'security' get through.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "\n",
    "\n",
    "class KeywordFilter(logging.Filter):\n",
    "    def __init__(self, keyword: str) -> None:\n",
    "        super().__init__()\n",
    "        self.keyword = keyword\n",
    "\n",
    "    def filter(self, record: logging.LogRecord) -> bool:\n",
    "        return self.keyword in record.getMessage()\n",
    "\n",
    "\n",
    "class ListHandler(logging.Handler):\n",
    "    def __init__(self, level: int = logging.NOTSET) -> None:\n",
    "        super().__init__(level)\n",
    "        self.messages: list[str] = []\n",
    "    def emit(self, record: logging.LogRecord) -> None:\n",
    "        self.messages.append(self.format(record))\n",
    "\n",
    "\n",
    "logger = logging.getLogger('filter-demo')\n",
    "logger.setLevel(logging.DEBUG)\n",
    "handler = ListHandler()\n",
    "handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))\n",
    "handler.addFilter(KeywordFilter('security'))\n",
    "logger.addHandler(handler)\n",
    "\n",
    "logger.info('user logged in')                         # filtered out\n",
    "logger.warning('security: failed login attempt')       # kept\n",
    "logger.info('data processed')                          # filtered out\n",
    "logger.error('security: unauthorised access attempt')  # kept\n",
    "\n",
    "for m in handler.messages:\n",
    "    print(' ', m)\n",
    "logger.handlers.clear()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Why this works\n",
    "\n",
    "Every handler inherits from `logging.Handler`, and the framework's whole job is to turn a `LogRecord` into a call to `emit(record)`. Level filtering, formatter application, thread safety, and locking all happen in the base class. You only need to decide what to do once the record arrives — so `emit` is usually three or four lines.\n",
    "\n",
    "`self.format(record)` does the formatter dance for you: applies the format string you set with `setFormatter`, renders `%(asctime)s`/`%(levelname)s`/etc., and returns a string. For destinations that want *structured* data rather than a rendered string, reach directly into the `record` attributes — `record.created`, `record.levelname`, `record.getMessage()`, `record.exc_info` — and skip `format()` entirely.\n",
    "\n",
    "Wrapping the body in `try` / `self.handleError(record)` is the standard defensive shape. If your CSV file is suddenly unwritable or the network handler times out, you want the failure logged by the framework rather than raised into the caller's code path.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Trade-offs\n",
    "\n",
    "Reach for a custom handler only when the built-ins really don't fit. `StreamHandler`, `FileHandler`, `RotatingFileHandler`, `TimedRotatingFileHandler`, `SMTPHandler`, and `SysLogHandler` between them cover most production needs. `MemoryHandler` even covers the in-process buffer case if you don't need an assertable list.\n",
    "\n",
    "For filtering by content (not level), add a `logging.Filter` rather than baking the logic into `emit`. Filters compose — one on the logger, one on each handler — and keep `emit` focused on the destination.\n",
    "\n",
    "When the handler does I/O that could block (HTTP, databases, slow disks), consider using `QueueHandler` on the app side with a background thread running the real handler. That way your request handler isn't waiting on the log system.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Related reading\n",
    "\n",
    "- [Configure logging for a project](https://agilearn.co.uk/guides/logging-and-debugging/recipes/configure-logging-for-a-project) — how to wire your custom handler into `dictConfig`.\n",
    "- [Avoid common logging mistakes](https://agilearn.co.uk/guides/logging-and-debugging/recipes/avoid-common-logging-mistakes) — handler-level vs. logger-level, duplicate handlers.\n",
    "- [Logging module quick reference](https://agilearn.co.uk/guides/logging-and-debugging/reference/logging-module-quick-reference) — the full list of built-in handlers.\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.12.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}