lace-app-sdk

Data collections

Collections are typed with JSON Schema and served through the governed AppDataService — tenant-scoped, audited, with an automatic admin panel and durable storage.

pythonapp/data.py

from lace_app_sdk.data import AppDataCollectionSchema, AppDataCapability, PhysicalStorageSpec, AppDataUniqueConstraint

TICKETS = AppDataCollectionSchema(
    app_id="acme.field_intake", collection_id="tickets", display_name="Tickets",
    storage_backend="generic_app_data",
    json_schema={"type": "object", "properties": {
        "subject": {"type": "string"},
        "priority": {"enum": ["low", "normal", "urgent"]},
        "evidence_blob": {"type": "string"},
    }},
    required_fields=["subject"], identity_fields=["subject"],
    unique_constraints=[AppDataUniqueConstraint(name="ux_tickets_subject", fields=["subject"])],
    searchable_fields=["subject"], sortable_fields=["subject", "updated_at"],
    description="Field intake tickets — one row per subject, searchable.",
    physical_storage=PhysicalStorageSpec(capabilities=[AppDataCapability.EVENT_LOG, AppDataCapability.FULL_TEXT_INDEX]),
)

Two planes, one rule

Dual-plane hard rule. Anything you list / get / query must be in a typed AppDataCollectionSchema (Plane 1). Use put_blob only for opaque/large payloads (Plane 2) and reference blob_ref from the row so queries remain indexed. Never use an in-memory dict/list for listing data — it dies on restart and GET /notes returns [] after redeploy (automatic proof-lane FAIL).

Plane	What it is	API
Plane 1 — RDS typed collections	JSONB rows in `app_data.records`, scoped by `(tenant_id, app_id, collection_id)`	`AppDataService.create_record / list_records / patch_record / delete_record`
Plane 2 — content-addressed blobs	MinIO/S3, deduped by hash	`put_blob / get_blob / put_blob_json / get_blob_json` (`lace_app_sdk.data` / `lace_app_sdk.blob`)

Schema fields that matter

Field	What it controls
`json_schema`	JSON Schema for the row; validated on write
`required_fields`	Must be present on create
`identity_fields`	Derives `record_id` (e.g. `["subject"]` → `record_id == subject`)
`unique_constraints`	DB-level UNIQUE → HTTP 409 `app_data_unique_constraint_failed` on violation
`searchable_fields` + `FULL_TEXT_INDEX`	GIN index for `search=` queries over those fields
`sortable_fields`	Allowed `sort_by`
`physical_storage.capabilities`	`EVENT_LOG` (audit row per write) + `FULL_TEXT_INDEX`

CRUD — the calls you actually make

pythonapp/routes.py — AppDataService

from lace_app_sdk.data import AppDataService, AppDataCreateRequest, AppDataQueryRequest, AppDataPatchRequest

svc = AppDataService(app_state=state)

# create — 409 on unique violation
rec = svc.create_record("acme.field_intake", "tickets",
                        AppDataCreateRequest(data_json={"subject": "Heater site 12", "priority": "urgent"}))
# → AppDataRecordEnvelope(record_id, data_json, created_at, updated_at)

# list / search — full-text over searchable_fields
page = svc.list_records("acme.field_intake", "tickets",
                         AppDataQueryRequest(search="heater", limit=20, offset=0))

# patch / delete
svc.patch_record("acme.field_intake", "tickets", rec.record_id,
                 AppDataPatchRequest(data_json={"priority": "normal"}))
svc.delete_record("acme.field_intake", "tickets", rec.record_id)

# blobs — opaque payloads, reference from row
from lace_app_sdk.data import put_blob, get_blob
blob_ref = put_blob(state, data=open("evidence.pdf","rb").read(), content_type="application/pdf")
# store blob_ref in row.evidence_blob so queries stay indexed

In AppRouteProvider.register_routes, register the collection once at boot (idempotent): AppDataService(state).register_collection(TICKETS). See src/lace_app_sdk/examples/notes_app/routes.py for the full GET /notes example (literal /notes/search before /notes/{note_id}).

Wiring to the manifest

Declare TICKETS in data_collections=[TICKETS] of your LaceAppManifest and call write_app_manifest(MANIFEST, "lace_app_manifest.json") so the publisher sees it. The publisher also auto-derives collections from app/data.py as a fallback, but explicit registration is preferred.

Admin panel & audit

Every collection gets a generated admin panel at /admin/app-data/{collection_id} — no code. Every create / patch / delete appends a row to the EVENT_LOG with tenant, principal, before/after, and trace id. Retention is per tenant (see governance).

Next: tools or routes & UI.