nikhilvdev · nikhilvdev · Aug 30, 2026 · Aug 30, 2026 · Aug 30, 2026
@@ -4,6 +4,46 @@ All notable changes to this project are documented in this file.
 
 ## Unreleased
 
+- New transports: SQL (`BaseSQLTransport` + `SQLiteTransport`,
+  `PostgresTransport`, `MySQLTransport`), NoSQL (`MongoDBTransport`,
+  `DynamoDBTransport`, `RedisTransport`), message queues
+  (`BaseQueueTransport` + `KafkaTransport`, `RabbitMQTransport`,
+  `SQSTransport`, `PubSubTransport`), and cloud-native sinks
+  (`CloudWatchTransport`, `CloudLoggingTransport`, `AppInsightsTransport`,
+  `DatadogTransport`, `ElasticsearchTransport`, `NewRelicTransport`) —
+  full parity with `logquill-js` 0.2.0. All of it sits on a new shared
+  `BatchingTransport` base that bounds its buffer by both record count and
+  estimated byte size, swaps the buffer out before sending so a
+  synchronous re-entrant flush can't double-send, and catches a failing
+  send rather than propagating it to the caller (logged via Python's
+  stdlib `logging.getLogger("logquill")`) — a slow or down sink can't
+  crash the process. Every optional backend driver (`psycopg2-binary`,
+  `pymysql`, `pymongo`, `boto3`, `redis`, `kafka-python`, `pika`,
+  `google-cloud-pubsub`, `google-cloud-logging`) is a lazy, injectable
+  dependency behind a new `pyproject.toml` extra (`postgres`, `mysql`,
+  `mongodb`, `redis`, `kafka`, `rabbitmq`, `pubsub`, `gcp-logging`, and a
+  shared `aws` extra for CloudWatch/DynamoDB/SQS, all boto3-backed); a
+  missing driver raises an actionable `ImportError` rather than a
+  cryptic one, and every test injects a hand-written fake instead of
+  requiring a live service. `SQLiteTransport` needs no extra at all
+  (stdlib `sqlite3`).
+
+  Two deliberate departures from `logquill-js`'s implementation, same
+  outward behavior: `AppInsightsTransport` posts to Application Insights'
+  public ingestion endpoint via stdlib `urllib` instead of an Azure SDK
+  dependency, and `SQSTransport` dispatches its 10-message chunks
+  sequentially rather than concurrently, since this project's dispatch is
+  still fully synchronous end to end (true concurrency arrives once a
+  non-blocking async worker exists). `SyslogTransport` isn't included
+  here either, matching `logquill-js` 0.2.0, which didn't ship it; it's a
+  shared follow-up for both packages, not a Python-only gap.
+
+  Also restructured `logquill/transport.py`, `console_transport.py`,
+  `file_transport.py`, and `http_transport.py` into a new
+  `logquill/transports/` subpackage (with `sql/`, `nosql/`, `queue/`, and
+  `cloud/` subpackages) to hold the 17 new transports — a pure move, the
+  public `from logquill import ...` surface is unchanged.
+
 - Plugin pipeline: `Plugin` base (`before_log`/`after_log`/`on_error`,
   all optional to override), `ContextPlugin` (merges fixed context into
   `meta`), `RedactPlugin` (replaces sensitive `meta` values by key, case-

@@ -22,7 +22,7 @@ landed so far.
 
 - **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](https://www.npmjs.com/package/logquill)
-- **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation), `HTTPTransport` (batched); write your own by subclassing `Transport`
+- **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
 - **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP
@@ -101,6 +101,152 @@ 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:
+
+```python
+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):
+
+```python
+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.
+
+| 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:
+
+```python
+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
+```
+
+| 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:
+
+```python
+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)
+```
+
+| 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

@@ -1,31 +1,70 @@
-from logquill.console_transport import ConsoleTransport
-from logquill.context_plugin import ContextPlugin
-from logquill.file_transport import FileTransport
 from logquill.formatter import Formatter, JSONFormatter
-from logquill.http_transport import HTTPTransport
 from logquill.levels import Level, parse_level
 from logquill.logger import Logger
-from logquill.plugin import Plugin
+from logquill.plugins.context_plugin import ContextPlugin
+from logquill.plugins.plugin import Plugin
+from logquill.plugins.redact_plugin import RedactPlugin
+from logquill.plugins.sampling_plugin import SamplingPlugin
 from logquill.records import LogRecord
-from logquill.redact_plugin import RedactPlugin
-from logquill.sampling_plugin import SamplingPlugin
-from logquill.transport import CollectingTransport, Transport
+from logquill.transports.batching_transport import BatchingTransport
+from logquill.transports.cloud.app_insights_transport import AppInsightsTransport
+from logquill.transports.cloud.cloud_logging_transport import CloudLoggingTransport
+from logquill.transports.cloud.cloudwatch_transport import CloudWatchTransport
+from logquill.transports.cloud.datadog_transport import DatadogTransport
+from logquill.transports.cloud.elasticsearch_transport import ElasticsearchTransport
+from logquill.transports.cloud.new_relic_transport import NewRelicTransport
+from logquill.transports.console_transport import ConsoleTransport
+from logquill.transports.file_transport import FileTransport
+from logquill.transports.http_transport import HTTPTransport
+from logquill.transports.nosql.dynamodb_transport import DynamoDBTransport
+from logquill.transports.nosql.mongodb_transport import MongoDBTransport
+from logquill.transports.nosql.redis_transport import RedisTransport
+from logquill.transports.queue.base_queue_transport import BaseQueueTransport
+from logquill.transports.queue.kafka_transport import KafkaTransport
+from logquill.transports.queue.pubsub_transport import PubSubTransport
+from logquill.transports.queue.rabbitmq_transport import RabbitMQTransport
+from logquill.transports.queue.sqs_transport import SQSTransport
+from logquill.transports.sql.base_sql_transport import BaseSQLTransport, SQLLogRow
+from logquill.transports.sql.mysql_transport import MySQLTransport
+from logquill.transports.sql.postgres_transport import PostgresTransport
+from logquill.transports.sql.sqlite_transport import SQLiteTransport
+from logquill.transports.transport import CollectingTransport, Transport
 
 __version__ = "0.1.3"
 
 __all__ = [
+    "AppInsightsTransport",
+    "BaseQueueTransport",
+    "BaseSQLTransport",
+    "BatchingTransport",
+    "CloudLoggingTransport",
+    "CloudWatchTransport",
     "CollectingTransport",
     "ConsoleTransport",
     "ContextPlugin",
+    "DatadogTransport",
+    "DynamoDBTransport",
+    "ElasticsearchTransport",
     "FileTransport",
     "Formatter",
     "HTTPTransport",
     "JSONFormatter",
+    "KafkaTransport",
     "Level",
     "LogRecord",
     "Logger",
+    "MongoDBTransport",
+    "MySQLTransport",
+    "NewRelicTransport",
     "Plugin",
+    "PostgresTransport",
+    "PubSubTransport",
+    "RabbitMQTransport",
     "RedactPlugin",
+    "RedisTransport",
+    "SQLLogRow",
+    "SQLiteTransport",
+    "SQSTransport",
     "SamplingPlugin",
     "Transport",
     "parse_level",

@@ -4,9 +4,9 @@
 from typing import Any
 
 from logquill.levels import Level, parse_level
-from logquill.plugin import Plugin
+from logquill.plugins.plugin import Plugin
 from logquill.records import LogRecord, create_record
-from logquill.transport import Transport
+from logquill.transports.transport import Transport
 
 
 class Logger:

@@ -2,7 +2,7 @@
 
 from typing import Any
 
-from logquill.plugin import Plugin
+from logquill.plugins.plugin import Plugin
 from logquill.records import LogRecord
 
 

@@ -2,7 +2,7 @@
 
 from collections.abc import Iterable
 
-from logquill.plugin import Plugin
+from logquill.plugins.plugin import Plugin
 from logquill.records import LogRecord
 
 DEFAULT_REDACTED_KEYS = frozenset({"password", "token", "secret", "api_key", "authorization"})

@@ -3,7 +3,7 @@
 import random
 from typing import Callable
 
-from logquill.plugin import Plugin
+from logquill.plugins.plugin import Plugin
 from logquill.records import LogRecord
 
 

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import json
+import logging
+from abc import abstractmethod
+from typing import Generic, Sequence, TypeVar, cast
+
+from logquill.formatter import Formatter
+from logquill.records import LogRecord
+from logquill.transports.transport import Transport
+
+T = TypeVar("T")
+
+_logger = logging.getLogger("logquill")
+
+
+class BatchingTransport(Transport, Generic[T]):
+    """Shared base for every transport that buffers records and sends them in
+    batches (SQL, NoSQL, message queue, and cloud-native sinks).
+
+    Bounds the buffer by **both** record count and estimated byte size —
+    count alone lets a handful of huge `meta` payloads blow past reasonable
+    memory before a batch triggers. A flush fires as soon as either bound is
+    hit, checked after every `write()`. `_send_batch` is never called with an
+    empty batch, and a failing send is caught and logged rather than
+    propagated — a slow or down sink can never crash the caller's process.
+    """
+
+    def __init__(
+        self,
+        *,
+        formatter: Formatter | None = None,
+        max_records: int = 100,
+        max_bytes: int = 1_000_000,
+    ) -> None:
+        super().__init__(formatter)
+        self.max_records = max_records
+        self.max_bytes = max_bytes
+        self._buffer: list[T] = []
+        self._buffer_bytes = 0
+
+    def _to_item(self, formatted: str, record: LogRecord) -> T:
+        return cast(T, record)
+
+    def _size_of(self, item: T) -> int:
+        return len(json.dumps(item, separators=(",", ":"), default=str).encode("utf-8"))
+
+    def write(self, formatted: str, record: LogRecord) -> None:
+        item = self._to_item(formatted, record)
+        self._buffer.append(item)
+        self._buffer_bytes += self._size_of(item)
+        if len(self._buffer) >= self.max_records or self._buffer_bytes >= self.max_bytes:
+            self.flush()
+
+    def flush(self) -> None:
+        if not self._buffer:
+            return
+        batch, self._buffer = self._buffer, []
+        self._buffer_bytes = 0
+        try:
+            self._send_batch(batch)
+        except Exception:
+            _logger.exception("%s: failed to send log batch", type(self).__name__)
+
+    def close(self) -> None:
+        self.flush()
+
+    @abstractmethod
+    def _send_batch(self, batch: Sequence[T]) -> None: ...