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

fromlogquillimportLevel, Loggerlogger=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 levellogger.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:

fromlogquillimportJSONFormatterprint(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):

fromlogquillimportConsoleTransport, FileTransport, HTTPTransport, Loggerlogger=Logger(
"app",
transports=[
ConsoleTransport(), # stdout, ERROR/FATAL to stderr, colorizedFileTransport("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:

fromlogquillimportCollectingTransport, Loggersink=CollectingTransport()
logger=Logger("app.test", transports=[sink])
logger.info("hello")
assertsink.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:

fromlogquillimportLogger, SQLiteTransporttransport=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):

fromlogquillimportLogger, MongoDBTransportclassFakeCollection:
def__init__(self):
self.documents= []
definsert_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)
assertcollection.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.

Transport	Driver	Extra
SQLiteTransport	stdlib sqlite3	(none)
PostgresTransport	psycopg2-binary	pip install logquill[postgres]
MySQLTransport	pymysql	pip install logquill[mysql]

NoSQL

Transport	Driver	Extra
MongoDBTransport	pymongo	pip install logquill[mongodb]
DynamoDBTransport	boto3	pip install logquill[aws]
RedisTransport	redis	pip install logquill[redis]

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:

fromlogquillimportLogger, SQSTransportclassFakeSQSClient:
def__init__(self):
self.calls= []
defsend_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])
foriinrange(12):
logger.info(f"event {i}")
# chunked into two send_message_batch calls: 10 messages, then 2

Transport	Driver	Extra
KafkaTransport	kafka-python	pip install logquill[kafka]
RabbitMQTransport	pika	pip install logquill[rabbitmq]
SQSTransport	boto3	pip install logquill[aws]
PubSubTransport	google-cloud-pubsub	pip install logquill[pubsub]

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:

fromlogquillimportDatadogTransport, LoggerclassFakeSender:
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)

Transport	Mechanism	Extra
CloudWatchTransport	boto3	pip install logquill[aws]
CloudLoggingTransport	google-cloud-logging	pip install logquill[gcp-logging]
AppInsightsTransport	stdlib urllib (public ingestion endpoint)	(none)
DatadogTransport	stdlib urllib	(none)
ElasticsearchTransport	stdlib urllib (_bulk API)	(none)
NewRelicTransport	stdlib urllib + gzip	(none)

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.

fromlogquillimportContextPlugin, Logger, RedactPlugin, SamplingPluginlogger=Logger("app")
logger.use(ContextPlugin(service="api", env="prod")) # merged into every record's metalogger.use(RedactPlugin(keys=["password", "token"])) # replaces matching meta valueslogger.use(SamplingPlugin(0.1)) # keep ~10% of records that reach this pointlogger.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:

fromlogquillimportLoggerdefstrip_ssn(record):
record["meta"].pop("ssn", None)
returnrecord# or None to drop the recordlogger=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.

fromlogquillimportCollectingTransport, Logger, SamplingPluginsink=CollectingTransport()
sampling=SamplingPlugin(0.01, transports=[sink]) # keep ~1%, tail-elevate the restlogger=Logger("app", transports=[sink], plugins=[sampling])
logger.info("received request", trace_id="req-42") # likely dropped — held in the bufferlogger.info("queried database", trace_id="req-42") # likely dropped — held in the bufferlogger.error("query timed out", trace_id="req-42") # elevates the whole traceassert [r["message"] forrinsink.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 metavalues — 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:

fromlogquillimportLogger, PIIRedactPluginlogger=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:

fromlogquillimportLogger, TamperEvidentPluginlogger=Logger("app", plugins=[TamperEvidentPlugin()])
records= [logger.info(f"step {i}") foriinrange(3)]
assertTamperEvidentPlugin.verify_chain(records) isTruerecords[1]["message"] ="tampered"# simulate an edited log lineassertTamperEvidentPlugin.verify_chain(records) isFalse

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:

fromlogquillimportLogger, PagerDutyAlertPlugin, SlackAlertPluginlogger=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 webhooklogger.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.