panini-connector 0.1.0

Drop-in HMAC-verified connector for receiving Panini SEO publish calls on self-hosted sites

panini-connector

Drop-in HMAC-verified connector for receiving Panini SEO publish calls on self-hosted sites. Turns 80 lines of copy-paste boilerplate into 5 lines.

Install

pip install "panini-connector[fastapi,supabase] @ git+https://github.com/prak16/PaniniOS.git#subdirectory=sdk/panini-connector-py"

(Or Flask / Postgres / all-of-the-above — see Extras below.)

Use — FastAPI + Supabase

from fastapi import FastAPI
from panini_connector import mount

app = FastAPI()
mount(
    app,
    path="/api/seo-connector",
    storage="supabase",
)

Set these env vars, then redeploy:

SEO_CONNECTOR_SECRET=<any long random string; must match Panini's config>
SUPABASE_URL=https://xxxxx.supabase.co
SUPABASE_SERVICE_ROLE_KEY=<service_role key from Supabase → Settings → API>

That's it. Panini will POST to /api/seo-connector when it publishes.

Use — Flask + Postgres

from flask import Flask
from panini_connector import mount_flask

app = Flask(__name__)
mount_flask(app, path="/api/seo-connector", storage="postgres")

Set:

SEO_CONNECTOR_SECRET=<...>
DATABASE_URL=postgresql://user:pass@host:5432/dbname

The SDK creates a panini_posts table on first call (idempotent). Skip this with mount_flask(..., storage_config={"auto_create_table": False}) and manage the schema yourself.

Use — Existing blog table (recommended for real sites)

If your site already has a blog_posts / posts / articles table your site renders from, don't create a new one — point the SDK at your existing table:

from fastapi import FastAPI
from panini_connector import mount
from panini_connector.storage.supabase import SupabaseStorage

app = FastAPI()
mount(
    app,
    path="/api/seo-connector",
    storage=SupabaseStorage(
        table="blog_posts",                              # your real table
        external_id_column="id",                         # your table's PK (DB-generated UUID)
        external_url_template="https://mysite.com/blog/{slug}",

        # Transform Panini post → your row. Rename columns, add extra static
        # columns, whatever your schema needs. This is where the mapping happens.
        to_row=lambda post: {
            "title": post["title"],
            "slug": post["slug"],
            "content": post["body_html"],                # renamed from body_html
            "description": post.get("excerpt") or post.get("seo_meta", {}).get("description", ""),
            "meta_title": post.get("seo_meta", {}).get("title") or post["title"],
            "meta_description": post.get("seo_meta", {}).get("description") or post.get("excerpt", ""),
            "featured_image_url": post.get("featured_image_url"),
            "author_name": post.get("author_name"),
            "status": post.get("status", "published"),
            # Extra columns your schema requires
            "category": "seo",
            "external_source": "panini",
        },
    ),
)

How the two knobs work:

• external_id_column — which column stores the ID you'll get back as external_id.

  • Set to "id" (or any name) when your table already has a DB-generated primary key. SDK inserts without an ID and reads it back from RETURNING id.
  • Leave unset (or "external_id") to have the SDK generate an ID (sb_a3f9…) and insert it. Requires that column to exist on your table.

• external_url_template — where posts render on your site. Substitutions: {slug}, {id}, {title}.

Row-Level Security note: if your existing table has RLS enabled, make sure your SUPABASE_SERVICE_ROLE_KEY can INSERT, UPDATE, DELETE on it. Or turn RLS off on that table — the service role bypasses RLS by default, but a badly configured policy can still block writes.

Use — Django + Supabase

Requires Django 4.1+ (async views). Add to your urls.py:

from django.urls import path
from panini_connector.framework.django import view

urlpatterns = [
    # ... your existing routes ...
    path(
        "api/seo-connector",
        view(storage="supabase"),
        name="panini_connector",
    ),
]

The returned view is CSRF-exempt (HMAC verification is our auth), so no extra middleware decorators needed. Set the same 3 env vars as the Supabase example above and redeploy.

Verify it works — the CLI

Ship it, then from anywhere:

panini-connector test \
    --url https://your-vm.example.com/api/seo-connector \
    --secret $SEO_CONNECTOR_SECRET

Output on success:

Testing connector at https://your-vm.example.com/api/seo-connector...
  ✓ ping        → 200 OK  site_name='My Blog'
  ✓ create_post → 200 OK  external_id='sb_a3f9b2c1'
  ✓ update_post → 200 OK
  ✓ delete_post → 200 OK

All ops working. Your connector is wired correctly.

If any step fails, the output tells you which op + the HTTP status

• the connector's error body — enough to debug.

What the SDK does under the hood

• Verifies X-Compass-Signature (HMAC-SHA256 over {timestamp}.{body})
• Rejects requests with timestamps outside a 5-min replay window
• Parses the versioned envelope
• Dispatches to your storage adapter (create_post / update_post / delete_post / ping)
• Returns {ok, external_id, external_url} per Panini's expected shape

Errors are surfaced with meaningful HTTP status codes:

• 401 — signature mismatch, missing headers, replay window expired
• 400 — malformed envelope, unsupported version
• 404 — update_post / delete_post with unknown external_id
• 500 — storage adapter raised an unexpected exception

Callbacks — run code after each publish

Fire a cache-invalidation, sitemap-regen, or Next.js revalidation call after each successful post change:

async def on_change(op, payload):
    if op in ("create_post", "update_post"):
        # e.g. next.js revalidation
        await httpx.post(f"https://mysite.com/api/revalidate?slug={payload['slug']}")

mount(app, on_post_change=on_change, storage="supabase")

Callback can be sync or async. Errors in the callback are logged but don't fail the request (the post is already persisted).

Custom storage

If you want to write to your own DB / CMS / API, implement StorageAdapter:

from panini_connector import mount, StorageAdapter

class MyStore:
    async def ping(self):
        return {"site_name": "My site", "version": "1.0"}
    async def create_post(self, post):
        # ... your insert code ...
        return {"external_id": "post_42", "external_url": "https://mysite.com/post/42"}
    async def update_post(self, external_id, post):
        # ... your update code ...
        return {"external_id": external_id, "external_url": "..."}
    async def delete_post(self, external_id):
        # ... your delete code ...
        pass

mount(app, storage=MyStore())

Extras

Base install has no dependencies. Framework + storage adapters are extras:

pip install panini-connector[fastapi]     # FastAPI mount
pip install panini-connector[flask]       # Flask mount
pip install panini-connector[supabase]    # Supabase storage
pip install panini-connector[postgres]    # Postgres storage (psycopg 3)
pip install panini-connector[cli]         # panini-connector CLI (httpx)
pip install panini-connector[all]         # everything

Envelope format

If you're implementing a custom StorageAdapter or debugging with curl, here's the exact request shape Panini sends:

POST /api/seo-connector
Content-Type: application/json
X-Compass-Timestamp: 1720000000
X-Compass-Signature: sha256=<64-hex>
X-Compass-Version: 1
X-Compass-Request-Id: <uuid>

{
  "op":         "create_post" | "update_post" | "delete_post" | "ping",
  "version":   "1",
  "request_id": "<uuid>",
  "timestamp": "2026-07-05T12:34:56Z",
  "post": {
    "title":              "...",
    "slug":               "...",
    "body_html":          "...",
    "excerpt":            "...",
    "categories":         [...],
    "tags":               [...],
    "featured_image_url": "...",
    "author_name":        "...",
    "seo_meta":           {...},
    "json_ld":            [...],
    "status":             "publish" | "draft"
  },
  "external_id": "..."   // only for update/delete
}

Expected response: HTTP 200 with {"ok": true, "external_id": "...", "external_url": "..."}

License

MIT.