fastapi-offline-sync

fastapi-offline-sync is a robust, production-ready offline-first sync layer for FastAPI applications backed by MongoDB. It provides high-performance synchronization endpoints and live WebSocket updates with built-in conflict resolution, multi-worker safety, and atomic transactions.

Features

• Incremental Pull: Efficient retrieval of incremental data changes using Hybrid Logical Clocks (HLC).
• Atomic Push Batches: Safely apply client mutations in atomic MongoDB transactions, preventing partial state corruption.
• WebSocket Streaming: Live Change Stream propagation for real-time document updates.
• Distributed Multi-Worker Safety: Custom HLC node namespaces derived automatically or configured via container metadata to eliminate collision risks.
• Soft-Delete Propagation: Seamless tombstones on full resyncs to clear obsolete client-side records.

Installation

pip install fastapi-offline-sync

Quick Start

Initialize and mount the router with production configurations:

import os
from fastapi import FastAPI
from fastapi_offline_sync import SyncConfig, SyncRouter

app = FastAPI(title="Production Sync Service")

config = SyncConfig(
    mongodb_uri=os.environ["MONGODB_URI"],
    database_name=os.environ["MONGODB_DATABASE"],
    collections=("tasks", "inventory_items"),
    # Set unique HLC node ID (e.g., container hostname / Pod IP) for multi-worker safety
    hlc_node_id=os.environ.get("HOSTNAME"),
)

app.include_router(SyncRouter(config))

Authentication & Identity Resolution

Configure a JWT dependency to verify credentials and return authenticated user properties.

from fastapi import Header, HTTPException, status

async def verify_jwt(
    authorization: str | None = Header(default=None),
) -> dict[str, str]:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing or invalid authentication header",
        )
    
    token = authorization.removeprefix("Bearer ")
    # Decode and verify the JWT with your auth provider
    # Extract identity fields:
    return {
        "user_id": "user-unique-identifier",
        "business_id": "business-tenant-identifier"
    }

Then register the dependency with SyncConfig:

config = SyncConfig(
    mongodb_uri=os.environ["MONGODB_URI"],
    database_name=os.environ["MONGODB_DATABASE"],
    collections=("tasks",),
    jwt_dependency=verify_jwt,
)

Fisco Integration

For Fisco-style backends where data is scoped by business/tenant rather than individual user, configure the sync engine to scope operations by business_id while keeping the acting identity as user_id.

from fastapi_offline_sync import SyncConfig, SyncRouter, SyncService

config = SyncConfig(
    mongodb_uri=os.environ["MONGODB_URI"],
    database_name="Fisco",
    collections=(
        "inventory_items",
        "categories",
        "orders",
        "customers",
        "sales",
    ),
    actor_id_field="user_id",
    scope_id_field="business_id",
    jwt_dependency=verify_jwt,
)

router = SyncRouter(config)
sync_service = SyncService(config)

Server-Layer Writes

For existing service-layer database writes, ensure that you append HLC metadata and oplog entries in the same session:

1. Retrieve sync metadata with sync_service.build_sync_metadata(actor=user, scope_id=business_id).
2. Persist the metadata fields _sync_version, _sync_actor_id, _sync_scope_id, and _sync_deleted with the document.
3. Record the change using sync_service.record_server_change(...).

Deploying to Production

Run the application using a production-grade ASGI server:

uvicorn examples.app:app --host 0.0.0.0 --port 8000 --workers 4

API Reference

1. Push Client Changes (POST /sync/push)

Pushes client-side mutations to the server.

curl -X POST https://api.yourservice.com/sync/push \
  -H 'Content-Type: application/json' \
  -H 'Authorization: Bearer <JWT_TOKEN>' \
  -d '{
    "client_id": "device-client-uuid",
    "changes": [
      {
        "collection": "tasks",
        "operation": "upsert",
        "doc_id": "task-abc-123",
        "data": {"title": "Acquire inventory", "done": false},
        "parent_version": "20260502T120000.000Z-0001-a1b2"
      }
    ]
  }'

2. Incremental Pull (GET /sync/pull)

Fetches operations since a specific HLC token.

curl -H 'Authorization: Bearer <JWT_TOKEN>' \
  'https://api.yourservice.com/sync/pull?since=20260502T120000.000Z-0001-a1b2&collections=tasks&limit=100'

3. Full Resync (GET /sync/full)

Triggers a complete sync with Gzip compression and tombstones for soft-deleted documents.

curl -H 'Authorization: Bearer <JWT_TOKEN>' \
  --compressed \
  'https://api.yourservice.com/sync/full?collections=tasks&limit=1000'

4. Live Updates WebSocket (WS /sync/stream)

Subscribes to live database modifications via a persistent WebSocket connection.

{
  "type": "subscribe",
  "since": "20260502T120000.000Z-0001-a1b2",
  "collections": ["tasks"],
  "token": "<JWT_TOKEN>"
}