Skip to content

connectors.md

Latest commit

 

History

History
247 lines (190 loc) · 8.57 KB

connectors.md

File metadata and controls

247 lines (190 loc) · 8.57 KB
Raw

Connectors — rag_backends.connectors

The Connector SPI (declared in rag_core.spi.connector) discovers documents in an external source and fetches their raw bytes. Three built-in implementations ship in rag-backends:

Connector Source Extra
FilesystemConnector Local directory tree —
S3Connector AWS S3, MinIO, any S3-compatible store (in rag-backends core)
GCSConnector Google Cloud Storage uv sync --extra gcs

Overview

A connector exposes two methods:

def list_documents(
    self,
    ctx: RequestContext,
    state: ConnectorState | None = None,
) -> AsyncIterator[tuple[Document, ConnectorState]]: ...

async def fetch(self, ctx: RequestContext, document: Document) -> bytes: ...

list_documents is a resumable async iterator: each yield is a (Document, ConnectorState) pair. The state carries an opaque cursor and last-seen wall clock; persisting the latest state lets a subsequent invocation resume an interrupted crawl without re-emitting earlier documents.

Usage

Filesystem

from rag_backends.connectors import FilesystemConnector
from rag_core.types import CorpusId, TenantId

connector = FilesystemConnector(
    root="/var/data/corpus",
    corpus_id=CorpusId("internal-docs"),
)
async for doc, state in connector.list_documents(ctx):
    raw = await connector.fetch(ctx, doc)
    ...  # parse, chunk, embed, index

S3 / MinIO

from rag_backends.connectors import S3Connector

connector = S3Connector(
    bucket="my-bucket",
    corpus_id=CorpusId("uploads"),
    prefix="2026/",
    endpoint_url="http://localhost:9000",  # MinIO in dev stack
    aws_access_key_id="minioadmin",
    aws_secret_access_key="minioadmin",
)

GCS

Requires the gcs extra:

uv sync --extra gcs
from rag_backends.connectors import GCSConnector

connector = GCSConnector(
    bucket="my-gcs-bucket",
    corpus_id=CorpusId("uploads"),
    prefix="2026/",
)

Credentials come from Application Default Credentials — set GOOGLE_APPLICATION_CREDENTIALS to a service-account key, or run on a GCE/GKE workload with an attached identity.

Resumable crawl

Persist the most-recently-emitted state and feed it back:

state: ConnectorState | None = await load_state(...)
async for doc, new_state in connector.list_documents(ctx, state):
    ...
    await save_state(new_state)

Each connector defines its own cursor semantics:

Connector Cursor
Filesystem Absolute path of the last-yielded file
S3 NextContinuationToken from ListObjectsV2 (per-page)
GCS nextPageToken from the GCS REST API

last_seen_at is also honoured — items older than the watermark are skipped.

Internals

The built-in connectors share the small Crawler base class in rag_backends.connectors._base. It handles the iteration loop, error policy ("fail" / "skip"), and monotonic-watermark bookkeeping. Subclasses plug in only discover() and to_document().

Crawler is intentionally internal: external connector authors implement Connector directly so the SPI remains the only stable surface.

Extension points

To add a new connector, implement rag_core.spi.connector.Connector and return (Document, ConnectorState) tuples whose ConnectorState.connector_id identifies the source unambiguously across tenants. See the architecture doc for the policy boundary and CDC interaction (Step 1.9).

CDC (incremental sync) — Step 1.9

Three additional connectors stream change events from live sources rather than walking a full listing. They use the same Connector SPI shape; the difference is that list_documents() blocks on the event stream and may run indefinitely.

Plugin Source Cursor Extra
PostgresCDCConnector Postgres logical replication via pgoutput WAL LSN [postgres-cdc]
S3EventsConnector S3 → SQS event notifications SQS receipt handle uses default [s3]
WebhookReceiverConnector inbound HTTP webhooks (gateway-driven) published-count none

PostgresCDCConnector

from rag_backends.connectors import PostgresCDCConnector

conn = PostgresCDCConnector(
    dsn="postgresql://user:pw@host/db",
    slot_name="rag_slot",
    publication="rag_pub",
    corpus_id=CorpusId("users"),
)
async for document, state in conn.list_documents(ctx):
    body = await conn.fetch(ctx, document)
    # body is the JSON-encoded row payload

Requires wal_level = logical on the server and a role with the REPLICATION attribute. The connector creates the slot on first attach (pass create_slot=False to require it pre-exist). INSERT and UPDATE events for the same primary key collapse to the same Document.id (pg:<table>:<pk>) so the latest value wins; DELETE events share the id and carry metadata["cdc_op"] == "D" so the ingest pipeline can treat them as tombstones.

S3EventsConnector

from rag_backends.connectors import S3EventsConnector

conn = S3EventsConnector(
    queue_url="https://sqs.us-east-1.amazonaws.com/123/rag-ingest",
    bucket="acme-corp-docs",
    corpus_id=CorpusId("corp-docs"),
)
async for document, state in conn.list_documents(ctx):
    body = await conn.fetch(ctx, document)

Configure the source bucket with event notifications targeting the SQS queue. The connector long-polls SQS, parses S3 notification payloads (handles both direct-to-SQS and SNS-wrapped envelopes), filters cross-bucket events, and emits one Document per record. Tombstones (ObjectRemoved) have metadata["s3_event"] == "ObjectRemoved"; fetch returns b"" for them since the object is gone.

WebhookReceiverConnector

from rag_backends.connectors import WebhookReceiverConnector

webhook = WebhookReceiverConnector(max_queue_size=1024, backpressure="wait")

# In the gateway's FastAPI route:
@app.post("/v1/webhooks/notion")
async def notion_webhook(payload: dict):
    document = build_document_from_notion(payload)
    await webhook.publish(document, content=payload["bytes"])

# In the ingest pipeline:
async for document, state in webhook.list_documents(ctx):
    body = await webhook.fetch(ctx, document)

Backpressure modes:

  • "wait" (default): producer awaits queue capacity.
  • "drop": producer logs and drops the event silently.
  • "raise": producer raises IngestionError so the HTTP route can return 503.

fetch() returns the bytes that were buffered by publish(). The buffer is process-local; persistent webhook state is the source system's responsibility (Notion / Slack / etc. retry on their own if the gateway is down).

DedupFilter — Step 1.9

from rag_backends.connectors import DedupFilter, S3Connector
from rag_backends import RedisCache

connector = DedupFilter(
    connector=S3Connector(bucket="docs", corpus_id=CorpusId("c1")),
    cache=RedisCache(url="redis://localhost:6379/0"),
    ttl_seconds=24 * 3600,
)
async for document, state in connector.list_documents(ctx):
    ...
print(connector.stats)  # DedupStats(seen=..., deduped=...)

Wraps any Connector and drops documents whose Document.fingerprint (== Document.content_hash) is already in the supplied cache. The cache key is namespaced by tenant_id so cross-tenant collisions are impossible. TTL bounds memory growth and reflects the dedup horizon — documents older than ttl_seconds are treated as fresh on next sight.

DedupFilter itself satisfies the Connector SPI; fetch() delegates to the wrapped connector since dedup is a list-time concern.

Errors

Symptom Cause / fix
IngestionError: root does not exist FilesystemConnector was constructed with a non-existent path.
IngestionError: cannot fetch non-file:// fetch() called with a Document whose source_uri does not match the connector's scheme.
ValueError: tenant_id ... does not match A resumed ConnectorState was constructed for a different tenant than the current ctx.
IngestionError: requires the [postgres-cdc] extra PostgresCDCConnector was used without pip install 'rag-backends[postgres-cdc]'.
IngestionError: queue is full WebhookReceiverConnector(backpressure="raise") hit max_queue_size.
IngestionError: no buffered content WebhookReceiverConnector.fetch() called for a document that wasn't published, or whose payload has been drained / evicted.
IngestionError: no buffered row PostgresCDCConnector.fetch() called for a document that wasn't emitted by the current list_documents() iteration.