Skip to content

Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nikhilvdev merged 1 commit into from
Aug 31, 2026
Merged

Add alerting, PII, tamper-evident & sampling #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
elevation — with `transports=` set, a record that would be dropped is
buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
discarded outright, and if any later record in that trace reaches
`elevate_at` (default `ERROR`), the whole trace — every buffered record
plus everything after — ships, flushed straight to `transports`. Buffering
is bounded by `max_buffered_records` and `max_traces`, oldest trace
evicted first. Without `transports`, behavior is unchanged from plain
rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
plain function alongside a `Plugin` instance — wrapped internally as an
anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
(emails, SSNs, credit-card numbers, phone numbers), recursing through
nested dicts/lists/tuples and matching regardless of which key holds the
value — complements `RedactPlugin`'s exact-key matching. Depth- and
cycle-bounded, so a circular reference or pathologically deep structure
can't hang or crash the caller. An opt-in `use_presidio=True` mode
(`pip install logquill[presidio]`) routes values through Microsoft
Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
record's own content plus the previous record's `meta.hash`, stored as
`meta.prev_hash`), so editing, removing, or reordering a line in a
written log breaks the chain from that point on. Ships with a static
`TamperEvidentPlugin.verify_chain(records)` to check a log after the
fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
`threshold`), with the actual send always running on a background
thread so a slow or unreachable destination can never block the log call
that triggered it. Repeated identical errors (same level + logger +
message by default, or a custom `dedupe_key`) within
`dedupe_window_seconds` collapse into one follow-up alert carrying an
occurrence count instead of spamming the destination once per record.
`send_alert` failures are caught and routed to the plugin's own
`on_error`, same as any other plugin hook. Tracking is bounded to
`max_tracked_keys` concurrent dedupe windows — alerting degrades under
extreme cardinality, logging itself never does. All three concrete
plugins use only the stdlib (`urllib`, `smtplib`) — no new required
dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
below): `Logger`'s per-transport dispatch had no error handling, so a
transport that failed to format or write a given record (e.g.
`JSONFormatter` on a `meta` value containing a circular reference) would
propagate the exception straight to the caller. Now caught and logged via
the same `logging.getLogger("logquill")` channel `BatchingTransport`
already uses, per transport, so one broken transport can't crash the
caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
`PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
deeply nested structures, unusual scalar types, non-JSON-serializable
values, and circular references — asserting the pipeline never crashes
the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
`PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
`DynamoDBTransport`, `RedisTransport`), message queues
Expand Down
125 changes: 123 additions & 2 deletions README.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
record["meta"].pop("ssn", None)
return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
"received request",
"queried database",
"query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
"app",
plugins=[
SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down
17 changes: 15 additions & 2 deletions logquill/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
"AlertingPlugin",
"AppInsightsTransport",
"BaseQueueTransport",
"BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
"DatadogTransport",
"DynamoDBTransport",
"ElasticsearchTransport",
"EmailAlertPlugin",
"FileTransport",
"Formatter",
"FunctionPlugin",
"HTTPTransport",
"JSONFormatter",
"KafkaTransport",
Expand All@@ -56,6 +65,8 @@
"MongoDBTransport",
"MySQLTransport",
"NewRelicTransport",
"PIIRedactPlugin",
"PagerDutyAlertPlugin",
"Plugin",
"PostgresTransport",
"PubSubTransport",
Expand All@@ -66,6 +77,8 @@
"SQLiteTransport",
"SQSTransport",
"SamplingPlugin",
"SlackAlertPlugin",
"TamperEvidentPlugin",
"Transport",
"parse_level",
"__version__",
Expand Down
30 changes: 24 additions & 6 deletions logquill/logger.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
def __init__(
self,
name: str,
level: int | str | Level = Level.INFO,
transports: list[Transport] | None = None,
plugins: list[Plugin] | None = None,
plugins: list[Plugin | MiddlewareFunc] | None = None,
) -> None:
self.name = name
self._level = parse_level(level)
self.transports: list[Transport] = list(transports) if transports else []
self.plugins: list[Plugin] = list(plugins) if plugins else []
self.plugins: list[Plugin] = []
for plugin in plugins or []:
self.use(plugin)

@property
def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
def set_level(self, level: int | str | Level) -> None:
self._level = parse_level(level)

def use(self, plugin: Plugin) -> Logger:
"""Register a plugin. Returns `self` so calls can be chained."""
def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
"""Register a plugin, or a plain `before_log`-style function.

A function is wrapped internally as an anonymous `Plugin`
(`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
without needing to read the `Plugin` base class first. Returns
`self` so calls can be chained.
"""
if not isinstance(plugin, Plugin):
plugin = FunctionPlugin(plugin)
self.plugins.append(plugin)
return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
record = result

for transport in self.transports:
transport.write(transport.format(record), record)
try:
transport.write(transport.format(record), record)
except Exception:
# a transport that can't format or write this particular record
# (e.g. a circular reference in `meta`) must not crash the caller
_logger.exception("%s: failed to write a log record", type(transport).__name__)

for plugin in self.plugins:
try:
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
Add alerting, PII, tamper-evident & sampling by nikhilvdev · Pull Request #6 · nikhilvdev/logquill-python · GitHub
Skip to content
Add alerting, PII, tamper-evident & sampling#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

nikhilvdev
merged 1 commit into

from

Aug 31, 2026
 Merged


Add alerting, PII, tamper-evident & sampling
#6
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


59 changes: 59 additions & 0 deletions
CHANGELOG.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,65 @@ All notable changes to this project are documented in this file.

## Unreleased

## 0.3.0 - 2026-08-31

- Plugin pipeline, Phase 4 complete: `SamplingPlugin` gained tail-based
 elevation — with `transports=` set, a record that would be dropped is
 buffered per `meta["trace_id"]` (configurable via `trace_key`) instead of
 discarded outright, and if any later record in that trace reaches
 `elevate_at` (default `ERROR`), the whole trace — every buffered record
 plus everything after — ships, flushed straight to `transports`. Buffering
 is bounded by `max_buffered_records` and `max_traces`, oldest trace
 evicted first. Without `transports`, behavior is unchanged from plain
 rate-based sampling.
- `Logger.use()` (and the `plugins=[...]` constructor list) now accepts a
 plain function alongside a `Plugin` instance — wrapped internally as an
 anonymous `Plugin` (`FunctionPlugin`) — so a one-off `before_log`-style
 transform doesn't require subclassing `Plugin` first.
- `PIIRedactPlugin`: regex-based PII redaction over `meta` **values**
 (emails, SSNs, credit-card numbers, phone numbers), recursing through
 nested dicts/lists/tuples and matching regardless of which key holds the
 value — complements `RedactPlugin`'s exact-key matching. Depth- and
 cycle-bounded, so a circular reference or pathologically deep structure
 can't hang or crash the caller. An opt-in `use_presidio=True` mode
 (`pip install logquill[presidio]`) routes values through Microsoft
 Presidio's analyzer/anonymizer instead, for ML-based detection; Presidio
 is imported lazily and stays a real, non-default dependency.
- `TamperEvidentPlugin`: hash-chains every record (`meta.hash` over the
 record's own content plus the previous record's `meta.hash`, stored as
 `meta.prev_hash`), so editing, removing, or reordering a line in a
 written log breaks the chain from that point on. Ships with a static
 `TamperEvidentPlugin.verify_chain(records)` to check a log after the
 fact. Opt-in — hashing every record has a real, measurable CPU cost.
- `AlertingPlugin` base class + `SlackAlertPlugin`, `PagerDutyAlertPlugin`,
 and `EmailAlertPlugin`: fires on ERROR/FATAL (or any configurable
 `threshold`), with the actual send always running on a background
 thread so a slow or unreachable destination can never block the log call
 that triggered it. Repeated identical errors (same level + logger +
 message by default, or a custom `dedupe_key`) within
 `dedupe_window_seconds` collapse into one follow-up alert carrying an
 occurrence count instead of spamming the destination once per record.
 `send_alert` failures are caught and routed to the plugin's own
 `on_error`, same as any other plugin hook. Tracking is bounded to
 `max_tracked_keys` concurrent dedupe windows — alerting degrades under
 extreme cardinality, logging itself never does. All three concrete
 plugins use only the stdlib (`urllib`, `smtplib`) — no new required
 dependency.
- Fixed a pre-existing gap surfaced by a new property-based test (see
 below): `Logger`'s per-transport dispatch had no error handling, so a
 transport that failed to format or write a given record (e.g.
 `JSONFormatter` on a `meta` value containing a circular reference) would
 propagate the exception straight to the caller. Now caught and logged via
 the same `logging.getLogger("logquill")` channel `BatchingTransport`
 already uses, per transport, so one broken transport can't crash the
 caller or stop other attached transports from receiving the record.
- Added a `hypothesis`-based property test (new `dev` dependency) that
 drives the plugin pipeline (`ContextPlugin`, `RedactPlugin`,
 `PIIRedactPlugin`, `TamperEvidentPlugin`) with adversarial `meta` —
 deeply nested structures, unusual scalar types, non-JSON-serializable
 values, and circular references — asserting the pipeline never crashes
 the caller, only ever fails closed.

- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
 `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
 `DynamoDBTransport`, `RedisTransport`), message queues
Expand Down

125 changes: 123 additions & 2 deletions
README.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@ landed so far.
- **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill)
- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport`
- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin`, `SamplingPlugin` out of the box; a broken plugin can't crash logging
- **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins))
- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
- **Typed throughout** — `mypy --strict` clean on the public API
- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md`
Expand DownExpand Up@@ -253,6 +253,9 @@ Plugins hook into the pipeline around each log call: `before_log(record)` can
transform a record or return `None` to drop it, `after_log(record)` runs once
it's been dispatched to every transport, and `on_error(exc, record)` catches
anything a plugin's own hooks raise — a broken plugin can't take down logging.
Records are **not** deep-copied through the pipeline — a plugin receives and
may mutate the same dict every other plugin sees; copy it yourself in
`before_log` if you need to preserve the original.

```python
from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin
Expand All@@ -267,7 +270,125 @@ logger.info("login attempt", user_id=42, password="hunter2")
# (unless this call was one of the ~90% sampling dropped, in which case it's None)
```

Write your own by subclassing `Plugin`; override only the hooks you need.
Write your own by subclassing `Plugin`; override only the hooks you need. For
a one-off transform, skip the subclass entirely — `.use()` also accepts a
plain function, wrapped internally as an anonymous `Plugin`:

```python
from logquill import Logger

def strip_ssn(record):
 record["meta"].pop("ssn", None)
 return record # or None to drop the record

logger = Logger("app")
logger.use(strip_ssn)
logger.info("submit", ssn="123-45-6789", user_id=42)
# meta: {'user_id': 42}
```

### Tail-based sampling elevation

Plain `SamplingPlugin(rate)` drops records independently of each other. Add
`transports=` and every record's `meta["trace_id"]` (configurable via
`trace_key`) turns sampling tail-based instead: a dropped record is buffered
under its trace id rather than discarded, and if any later record in that
same trace reaches `elevate_at` (default `ERROR`), the whole trace — every
buffered record plus everything from then on — ships, flushed straight to
`transports`. A request that looked unremarkable when it started still
produces a complete trace once it turns out to have failed.

```python
from logquill import CollectingTransport, Logger, SamplingPlugin

sink = CollectingTransport()
sampling = SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the rest
logger = Logger("app", transports=[sink], plugins=[sampling])

logger.info("received request", trace_id="req-42") # likely dropped — held in the buffer
logger.info("queried database", trace_id="req-42") # likely dropped — held in the buffer
logger.error("query timed out", trace_id="req-42") # elevates the whole trace

assert [r["message"] for r in sink.records] == [
 "received request",
 "queried database",
 "query timed out",
]
```

Buffering is bounded by `max_buffered_records` and `max_traces` — the oldest
buffered trace is evicted once either limit is hit, so a single
high-cardinality or long-lived trace can't grow memory without limit.

### PII redaction by pattern, not just key

`RedactPlugin` redacts by exact key match. `PIIRedactPlugin` complements it by
scanning `meta` **values** — recursively through nested dicts/lists/tuples —
for emails, SSNs, credit-card numbers, and phone numbers, and redacts matches
wherever they appear, regardless of which key holds them:

```python
from logquill import Logger, PIIRedactPlugin

logger = Logger("app", plugins=[PIIRedactPlugin()])

logger.info("support ticket", notes="reach me at jane@example.com, ssn 123-45-6789")
# meta: {'notes': 'reach me at ***, ssn ***'}
```

Detection is regex-based by default — fast, dependency-free, matched on shape
rather than meaning. For fuzzier ML-based detection instead, pass
`use_presidio=True` (`pip install logquill[presidio]`) to route values
through Microsoft Presidio's analyzer/anonymizer; Presidio stays a real,
opt-in dependency, never a default one.

### Tamper-evident logs

`TamperEvidentPlugin` hash-chains every record — each one's `meta.hash` covers
its own content plus the previous record's hash — so editing, removing, or
reordering a line in a written log breaks the chain from that point on.
Opt-in, since hashing every record has a real CPU cost:

```python
from logquill import Logger, TamperEvidentPlugin

logger = Logger("app", plugins=[TamperEvidentPlugin()])
records = [logger.info(f"step {i}") for i in range(3)]

assert TamperEvidentPlugin.verify_chain(records) is True

records[1]["message"] = "tampered" # simulate an edited log line
assert TamperEvidentPlugin.verify_chain(records) is False
```

### Alerting on errors

`AlertingPlugin` is a base class for firing an external alert on ERROR/FATAL
(or any configurable `threshold`). It never blocks the log call that
triggered it — the actual send runs on a background thread — and repeated
identical errors within `dedupe_window_seconds` collapse into a single
follow-up alert carrying an occurrence count, instead of spamming the
destination once per record. Concrete subclasses ship for Slack, PagerDuty,
and email:

```python
from logquill import Logger, PagerDutyAlertPlugin, SlackAlertPlugin

logger = Logger(
 "app",
 plugins=[
 SlackAlertPlugin("https://hooks.slack.com/services/T000/B000/xxx"),
 PagerDutyAlertPlugin("your-events-api-v2-routing-key", threshold="FATAL"),
 ],
)

logger.error("payment webhook failed") # posts to the Slack webhook
logger.fatal("database unreachable") # also pages via PagerDuty (threshold=FATAL)
```

Write your own destination by subclassing `AlertingPlugin` and implementing
`send_alert(record, occurrences)`; thresholding, deduplication, and the
never-block-the-caller behavior are all handled by the base class.

## Development

Expand Down

17 changes: 15 additions & 2 deletions
logquill/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
from logquill.formatter import Formatter, JSONFormatter
from logquill.levels import Level, parse_level
from logquill.logger import Logger
from logquill.plugins.alerting_plugin import AlertingPlugin
from logquill.plugins.context_plugin import ContextPlugin
from logquill.plugins.plugin import Plugin
from logquill.plugins.email_alert_plugin import EmailAlertPlugin
from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin
from logquill.plugins.pii_redact_plugin import PIIRedactPlugin
from logquill.plugins.plugin import FunctionPlugin, Plugin
from logquill.plugins.redact_plugin import RedactPlugin
from logquill.plugins.sampling_plugin import SamplingPlugin
from logquill.plugins.slack_alert_plugin import SlackAlertPlugin
from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin
from logquill.records import LogRecord
from logquill.transports.batching_transport import BatchingTransport
from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
Expand All@@ -30,9 +36,10 @@
from logquill.transports.sql.sqlite_transport import SQLiteTransport
from logquill.transports.transport import CollectingTransport, Transport

__version__ = "0.2.1"
__version__ = "0.3.0"

__all__ = [
 "AlertingPlugin",
 "AppInsightsTransport",
 "BaseQueueTransport",
 "BaseSQLTransport",
Expand All@@ -45,8 +52,10 @@
 "DatadogTransport",
 "DynamoDBTransport",
 "ElasticsearchTransport",
 "EmailAlertPlugin",
 "FileTransport",
 "Formatter",
 "FunctionPlugin",
 "HTTPTransport",
 "JSONFormatter",
 "KafkaTransport",
Expand All@@ -56,6 +65,8 @@
 "MongoDBTransport",
 "MySQLTransport",
 "NewRelicTransport",
 "PIIRedactPlugin",
 "PagerDutyAlertPlugin",
 "Plugin",
 "PostgresTransport",
 "PubSubTransport",
Expand All@@ -66,6 +77,8 @@
 "SQLiteTransport",
 "SQSTransport",
 "SamplingPlugin",
 "SlackAlertPlugin",
 "TamperEvidentPlugin",
 "Transport",
 "parse_level",
 "__version__",
Expand Down

30 changes: 24 additions & 6 deletions
logquill/logger.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,26 +1,31 @@
from __future__ import annotations

import contextlib
import logging
from typing import Any

from logquill.levels import Level, parse_level
from logquill.plugins.plugin import Plugin
from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin
from logquill.records import LogRecord, create_record
from logquill.transports.transport import Transport

_logger = logging.getLogger("logquill")


class Logger:
 def __init__(
 self,
 name: str,
 level: int | str | Level = Level.INFO,
 transports: list[Transport] | None = None,
 plugins: list[Plugin] | None = None,
 plugins: list[Plugin | MiddlewareFunc] | None = None,
 ) -> None:
 self.name = name
 self._level = parse_level(level)
 self.transports: list[Transport] = list(transports) if transports else []
 self.plugins: list[Plugin] = list(plugins) if plugins else []
 self.plugins: list[Plugin] = []
 for plugin in plugins or []:
 self.use(plugin)

 @property
 def level(self) -> Level:
Expand All@@ -29,8 +34,16 @@ def level(self) -> Level:
 def set_level(self, level: int | str | Level) -> None:
 self._level = parse_level(level)

 def use(self, plugin: Plugin) -> Logger:
 """Register a plugin. Returns `self` so calls can be chained."""
 def use(self, plugin: Plugin | MiddlewareFunc) -> Logger:
 """Register a plugin, or a plain `before_log`-style function.

 A function is wrapped internally as an anonymous `Plugin`
 (`FunctionPlugin`) — the same middleware ergonomics as Express/Koa,
 without needing to read the `Plugin` base class first. Returns
 `self` so calls can be chained.
 """
 if not isinstance(plugin, Plugin):
 plugin = FunctionPlugin(plugin)
 self.plugins.append(plugin)
 return self

Expand DownExpand Up@@ -60,7 +73,12 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord |
 record = result

 for transport in self.transports:
 transport.write(transport.format(record), record)
 try:
 transport.write(transport.format(record), record)
 except Exception:
 # a transport that can't format or write this particular record
 # (e.g. a circular reference in `meta`) must not crash the caller
 _logger.exception("%s: failed to write a log record", type(transport).__name__)

 for plugin in self.plugins:
 try:
Expand Down
Loading