logquill

A structured, leveled logging framework for Python with pluggable transports. Sibling to logquill on npm (logquill-js) — same log record shape, same level names, one mental model across a Python + Node stack.

Status: pre-release, under active development. The core Logger, level filtering, transports, and the plugin pipeline are implemented; non-blocking async dispatch is not yet — see CHANGELOG.md for what's landed so far.

Features

• Structured by default — every call carries a meta dict, not just a message string
• Cross-language record shape — identical JSON shape and level names/weights as logquill on npm
• Pluggable transports — ConsoleTransport (colorized, stderr for errors), FileTransport (rotation), HTTPTransport (batched), plus SQL/NoSQL/message-queue/cloud-native sinks (see 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 (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)
• 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

Install

pip install logquill

Quickstart

from logquill import Level, Logger

logger = Logger("app", level=Level.INFO)

record = logger.info("user signed up", user_id=42, plan="pro")
print(record)
# {'timestamp': '2026-08-27T18:04:12.345Z', 'level': 'INFO', 'logger': 'app',
#  'message': 'user signed up', 'meta': {'user_id': 42, 'plan': 'pro'}}

logger.debug("below threshold, dropped")  # -> None, filtered by level
logger.set_level("debug")
logger.debug("now visible")  # -> a record dict

Every log call returns the record dict (or None if filtered by level) — {"timestamp": ISO8601, "level": str, "logger": str, "message": str, "meta": dict}, the same shape shared with logquill on npm. Use JSONFormatter to serialize a record to the canonical JSON line:

from logquill import JSONFormatter

print(JSONFormatter().format(record))
# '{"timestamp":"2026-08-27T18:04:12.345Z","level":"INFO","logger":"app","message":"user signed up","meta":{"user_id":42,"plan":"pro"}}'

Transports

Attach transports to a Logger to actually write records somewhere. Each record is dispatched to every attached transport synchronously (non-blocking dispatch isn't implemented yet):

from logquill import ConsoleTransport, FileTransport, HTTPTransport, Logger

logger = Logger(
    "app",
    transports=[
        ConsoleTransport(),  # stdout, ERROR/FATAL to stderr, colorized
        FileTransport("app.log", max_bytes=10 * 1024 * 1024, backup_count=5),
        HTTPTransport("https://logs.example.com/ingest", batch_size=50),
    ],
)

logger.info("user signed up", user_id=42, plan="pro")
logger.close()  # flushes the file handle and any buffered HTTP batch

Write your own transport by subclassing Transport and implementing write(formatted, record); format(record) and close() have sensible defaults. CollectingTransport is a ready-made in-memory transport, handy in your own tests:

from logquill import CollectingTransport, Logger

sink = CollectingTransport()
logger = Logger("app.test", transports=[sink])

logger.info("hello")
assert sink.records[0]["message"] == "hello"

SQL, NoSQL, message queue, and cloud-native transports

Every transport below shares one design: records are always batched (bounded by both count and estimated byte size via a shared BatchingTransport base — never one write per log call), and every optional backend driver is a lazy, injectable dependency — pass a pre-built client/connection for tests or an alternate setup, or let the transport construct one itself from the real driver on first use. A missing driver raises an actionable ImportError telling you which extra to install, the same shape every transport in this list follows.

SQLiteTransport needs no optional dependency at all (stdlib sqlite3), so it's fully runnable as-is:

from logquill import Logger, SQLiteTransport

transport = SQLiteTransport(filename="app.db", ensure_schema=True, max_records=100)
logger = Logger("app", transports=[transport])

logger.info("user signed up", user_id=42, run_id="run-1")
logger.close()  # flushes any buffered rows

Every other backend follows the same injection shape — here's MongoDBTransport with a hand-rolled fake standing in for a real pymongo collection (the same pattern every transport's own test suite uses, so you never need a live service to test your own logging setup):

from logquill import Logger, MongoDBTransport

class FakeCollection:
    def __init__(self):
        self.documents = []
    def insert_many(self, documents):
        self.documents.extend(documents)

collection = FakeCollection()
transport = MongoDBTransport(collection=collection, max_records=1)
logger = Logger("app", transports=[transport])

logger.info("user signed up", user_id=42)
assert collection.documents[0]["message"] == "user signed up"

Passing a real pymongo.Collection instead of a fake works identically — MongoDBTransport(uri="mongodb://localhost:27017", database="app", collection_name="logs") builds one lazily via the optional pymongo peer dependency.

SQL — BaseSQLTransport (a fixed logs table: timestamp/level/ logger/message/meta, plus run_id/span_id/parent_span_id/ trace_id for upcoming cross-service trace-correlation support). ensure_schema=True is a dev/test convenience only — production schema/migrations are your responsibility, same as every batching transport below.

NoSQL

DynamoDBTransport partitions by meta["run_id"] (falling back to meta["trace_id"], then the logger name) with timestamp as the sort key. RedisTransport writes to a Redis Stream via XADD — a fast local buffer/tail, not a durable store.

Message queues — BaseQueueTransport (topic names the Kafka topic / RabbitMQ queue / SQS queue URL / GCP Pub/Sub topic path). Decouples log producers from consumers so a SIEM, an analytics pipeline, and an alerting system can all fan out from one topic. SQSTransport chunks at the API's 10-message SendMessageBatch cap:

from logquill import Logger, SQSTransport

class FakeSQSClient:
    def __init__(self):
        self.calls = []
    def send_message_batch(self, QueueUrl, Entries):
        self.calls.append((QueueUrl, Entries))

client = FakeSQSClient()
transport = SQSTransport(
    topic="https://sqs.us-east-1.amazonaws.com/123456789012/app-logs",
    client=client,
    max_records=12,
)
logger = Logger("app", transports=[transport])
for i in range(12):
    logger.info(f"event {i}")
# chunked into two send_message_batch calls: 10 messages, then 2

Cloud-native — DatadogTransport, ElasticsearchTransport, and AppInsightsTransport need no client SDK at all: each POSTs directly to its provider's public ingestion endpoint via stdlib urllib, with an injectable sender for tests:

from logquill import DatadogTransport, Logger

class FakeSender:
    def __init__(self):
        self.calls = []
    def __call__(self, url, api_key, batch):
        self.calls.append((url, api_key, batch))

sender = FakeSender()
transport = DatadogTransport(api_key="dd-api-key", sender=sender, max_records=1)
logger = Logger("app", transports=[transport])

logger.info("user signed up", user_id=42)

NewRelicTransport gzips every payload, strips meta["eventType"] (New Relic's reserved key), and on a 429 response reads Retry-After and pauses sends until it elapses — dropping (not requeuing) any batch flushed during that window, since New Relic blocks the rest of that minute on a rate-limit breach anyway.

Plugins

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.

from logquill import ContextPlugin, Logger, RedactPlugin, SamplingPlugin

logger = Logger("app")
logger.use(ContextPlugin(service="api", env="prod"))  # merged into every record's meta
logger.use(RedactPlugin(keys=["password", "token"]))  # replaces matching meta values
logger.use(SamplingPlugin(0.1))  # keep ~10% of records that reach this point

logger.info("login attempt", user_id=42, password="hunter2")
# meta: {'service': 'api', 'env': 'prod', 'user_id': 42, 'password': '***'}
# (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. For a one-off transform, skip the subclass entirely — .use() also accepts a plain function, wrapped internally as an anonymous Plugin:

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.

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:

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:

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:

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

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev,http,hooks]"
pre-commit install

ruff check .
mypy logquill
pytest

See CONTRIBUTING.md for the PR workflow, the Code of Conduct for community standards, and SECURITY.md for how to report a vulnerability.