Django connector

A pip-installable Django app that streams a site's model content to the bAInquet ingestion API, mapping each registered model to the canonical IngestItem shape, signing every request with the bq.connector.hmac.v1 HMAC scheme, and POSTing idempotent items on save and delete plus full batches on backfill.

Stable HMAC parity-tested: the signer emits the exact known-good X-Signature, byte-identical to the Node and PHP signers.

This is the only Python bAInquet connector, and it has zero third-party runtime dependencies: the signer uses the standard library (hashlib plus hmac) and the HTTP client uses urllib. No requests, no cryptography.

What it maps

Each registered instance maps to an IngestItem:

• type is your mapper's source_type, a bAInquet SourceType (page, post, product, category, collection, faq, review, event, location, profile, doc, media, custom).
• id is a stable "<type>:<pk>" (for example post:412), so re-ingest is idempotent.
• url and canonical_url are absolute on BAINQUET_SITE_DOMAIN; the server requires the url host to equal X-Site-Domain.
• title, html, text (derived from html), language, updated_at (RFC3339 UTC Z).
• json holds your model-specific structured data; the bundled ProductMapper maps price as a typed { value, unit: "currency", currency } object, never a string like "$19.99".
• checksum is a deterministic sha256:<hex> over the content-bearing fields (it excludes the volatile updated_at), so an unchanged instance produces the same checksum and the server returns skipped.

Install

pip install bainquet-connector

Add it to INSTALLED_APPS, after your content apps:

# settings.py
INSTALLED_APPS = [
    # ... your content apps first ...
    "bainquet_connector",
]

Configuration

# settings.py
import os

BAINQUET_CONNECTOR_TOKEN = os.environ["BAINQUET_CONNECTOR_TOKEN"]  # "connectorId.secret"
BAINQUET_WEBSITE_ID      = os.environ["BAINQUET_WEBSITE_ID"]        # token-scope website id
BAINQUET_SITE_DOMAIN     = "example.com"                            # X-Site-Domain (no scheme)
BAINQUET_BASE_URL        = "https://api.bainquet.online/v1/ingest"  # default; override for sandbox
BAINQUET_CONNECTOR_VERSION = "1.0.0"                               # advertised X-Connector-Version
BAINQUET_ENABLED         = True                                     # master switch

Setting	Required	Default	Purpose
BAINQUET_CONNECTOR_TOKEN	yes	—	connectorId.secret from your dashboard. Shown once.
BAINQUET_WEBSITE_ID	yes	—	Website id the token is scoped to (the HKDF salt).
BAINQUET_SITE_DOMAIN	yes	—	X-Site-Domain. Every emitted url host must equal this.
BAINQUET_BASE_URL	no	https://api.bainquet.online/v1/ingest	Ingestion API base.
BAINQUET_CONNECTOR_VERSION	no	1.0.0	X-Connector-Version.
BAINQUET_ENABLED	no	True	When False, the connector no-ops and sends nothing.

The token and secret are read only from settings; the package stores nothing and never logs them (a redactor masks token, secret, X-Signature, and Authorization).

Mapping pattern: register models (default-deny)

Nothing is ingested until you register a mapper per model. A model is ingestable only if it is registered: signals fire and backfill iterates over registered models only.

Register in your host app's AppConfig.ready(), so registration happens at startup before the connector connects its signals:

# myblog/apps.py
from django.apps import AppConfig

class MyBlogConfig(AppConfig):
    name = "myblog"

    def ready(self):
        from bainquet_connector.registry import register
        from bainquet_connector.examples import BlogPostMapper, ProductMapper
        from myblog.models import Post
        from myshop.models import Product

        register(Post, BlogPostMapper)
        register(Product, ProductMapper)

TIP

Place bainquet_connector after your content apps in INSTALLED_APPS so your ready() (which registers mappers) runs before the connector's ready() (which connects signals). If you register later, call bainquet_connector.signals.connect(BainquetConfig.from_settings()) again; it is idempotent.

Writing a custom mapper

Subclass BainquetMapper for full control; the base class builds the stable id, the absolute url, derived plain text, the RFC3339 updated_at, the deterministic checksum, and the one-of html/text/json enforcement:

from bainquet_connector.mapper import BainquetMapper, html_to_text

class ArticleMapper(BainquetMapper):
    source_type = "post"          # a SourceType enum value

    def get_title(self, instance):
        return instance.headline

    def get_html(self, instance):
        return instance.body_html

    def get_json(self, instance):
        # Model-specific data nests here, never as top-level IngestItem fields.
        return {"article": {"section": instance.section,
                            "tags": list(instance.tags.values_list("name", flat=True))}}

Incremental sync

The connector connects Django signals on registered models:

• post_save maps the instance and POSTs it to POST /v1/ingest/item.
• post_delete POSTs a tombstone to POST /v1/ingest/delete with the instance's stable id.

An ingest failure is logged (redacted) and never breaks your save or delete.

Backfill

python manage.py bainquet_backfill                       # all registered models
python manage.py bainquet_backfill --model myblog.Post   # one model
python manage.py bainquet_backfill --chunk 200           # chunk size (max 500)
python manage.py bainquet_backfill --dry-run             # map + validate locally, send nothing
python manage.py bainquet_backfill --limit 10            # cap per model (smoke test)

bainquet_backfill maps every registered instance and batch-POSTs to POST /v1/ingest/batch in chunks of at most 500 items.

Signing

signingKey  = HKDF-SHA256(secret, salt = websiteId, info = "bq.connector.hmac.v1", 32 bytes)  # RFC 5869, stdlib only
bodySha256  = sha256(body)  # over the exact JSON bytes sent
canonical   = METHOD\npath\nbodySha256\ntimestamp\nnonce\nwebsiteId
X-Signature = hex HMAC-SHA256(signingKey, canonical)

The body is serialized once, then hashed, signed, and sent; it is never re-serialized. Retries on 429 / 5xx reuse the same Idempotency-Key with a fresh nonce and timestamp. Headers sent: Authorization: Bearer <token>, Content-Type: application/json, X-Site-Domain, X-Connector-Type: django, X-Connector-Version, X-Signature, X-Timestamp, X-Nonce, X-Body-Sha256, Idempotency-Key.

See Ingestion and signing for the full scheme.

Out of scope

• The connector ingests only registered models, returning only the records your querysets expose. Use --dry-run to verify exactly what would be sent.

Verifying the signer

cd connectors/django/bainquet_connector
python -m unittest tests.test_signer_parity -v   # HMAC parity vs the known-good vector (no Django)
python -m unittest tests.test_mapper_unit -v     # mapper pure-transform unit tests (no Django)