Forking this into your product

The pattern is two halves that share a contract. You will (1) describe your entities as EntitySpecs, (2) write a CrudStore adapter over your real database, (3) wire tenant resolution and (optionally) an audit emitter, and (4) point the frontend store factory at your routes. Nothing here references a domain, so there is no vocabulary to strip — only seams to fill.

1. Describe your entity

from vue_pinia_crud_store import EntitySpec

ORDER_SPEC = EntitySpec(
    resource="orders",                # URL segment + audit entity_type + list key
    table_name="orders",              # SQL table (often == resource)
    pk_column="order_id",
    columns=(
        "order_id", "reference", "status", "total", "metadata",
        "created_at", "updated_at",
    ),
    filterable_columns={"reference", "status", "total", "created_at"},
    patchable_columns={"reference", "status", "total", "metadata"},
    required_create_columns=("reference",),
    jsonb_columns=("metadata",),
    has_updated_at=True,              # False for append-only / event tables
)

If you omit filterable_columns it defaults to all columns; if you omit patchable_columns it defaults to columns minus pk_column / created_at / updated_at. The two example specs (WIDGET_SPEC, EVENT_SPEC) are concrete references — copy their shape, not their names.

2. Implement the storage seam

CrudStore is a small duck-typed interface. The shipped InMemoryStore is the dependency-free default (great for tests and a local frontend harness). For production, write an adapter over your driver. A sketch over asyncpg:

from vue_pinia_crud_store import CrudStore, ListPage, FilterSpec, compile_filters
from vue_pinia_crud_store import encode_cursor, decode_cursor

class PgStore(CrudStore):
    def __init__(self, pool, spec):
        self.pool, self.spec = pool, spec

    async def list_rows(self, tenant_id, *, filters, allowed_columns, limit, cursor):
        where, params = compile_filters(filters, allowed_columns, table_alias="t")
        keyset = decode_cursor(cursor)
        clause = ""
        if keyset is not None:
            ts, rid = keyset
            params += [ts, rid]
            clause = f" AND (t.created_at, t.{self.spec.pk_column}) < (${len(params)-1}, ${len(params)})"
        params.append(limit + 1)
        sql = (
            f"SELECT {', '.join(self.spec.columns)} FROM {self.spec.table_name} t "
            f"WHERE {where}{clause} "
            f"ORDER BY t.created_at DESC, t.{self.spec.pk_column} DESC LIMIT ${len(params)}"
        )
        async with self.pool.acquire() as conn:
            await conn.execute("SET LOCAL app.current_tenant_id = $1", tenant_id)  # RLS
            rows = [dict(r) for r in await conn.fetch(sql, *params)]
        has_more = len(rows) > limit
        rows = rows[:limit]
        nxt = encode_cursor(rows[-1]["created_at"], str(rows[-1][self.spec.pk_column])) if has_more and rows else None
        return ListPage(rows=rows, next_cursor=nxt)

    # get_row / create_row / patch_row / delete_row / bulk: same shape, using
    # build_bulk_statement(...) for the bulk path.

Tenant isolation can be explicit (filter by tenant_id) or delegated to row-level security with a per-connection SET LOCAL (as above). The seam only requires that one tenant never sees another's rows.

3. Wire the router

import uuid
from fastapi import Depends, FastAPI
from vue_pinia_crud_store import build_crud_router

async def current_session(authorization: str = ...):
    return verify(authorization)        # your auth; returns an object/dict

async def audit_emitter(tenant_id, *, entity_type, operation, entity_id, diff, bulk_count):
    await write_ledger_row(tenant_id, entity_type, operation, entity_id, diff, bulk_count)

app = FastAPI()
app.include_router(build_crud_router(
    ORDER_SPEC, PgStore(pool, ORDER_SPEC),
    tenant_resolver=lambda session: session["tenant_id"],
    auth_dependency=current_session,    # omit → demo X-Tenant-Id header
    audit_emitter=audit_emitter,        # omit → mutations not audited
    id_validator=uuid.UUID,             # omit → ids passed through as strings
))

Repeat include_router(build_crud_router(SPEC, store, ...)) per entity.

4. Point the frontend store at your routes

import { createCrudStore } from 'vue-pinia-crud-store/createCrudStore';

export const useOrders = createCrudStore({
  resource: 'orders',
  idField: 'order_id',         // match the backend pk_column
  baseUrl: '/api',
  headers: { Authorization: `Bearer ${token}` },
  // or pass `http` to use an axios-shaped client / inject auth refresh
});

5. Use the pieces independently

You don't have to take the whole stack:

Use compile_filters(specs, allowed_columns, table_alias=...) as a safe filter→SQL compiler in a query layer you already have.
Use encode_cursor / decode_cursor for keyset pagination anywhere.
Use build_bulk_statement(...) to get a parameterized bulk statement without the router.
Use CrudHandlers(spec, store) for a framework-free CRUD surface (e.g. behind a GraphQL resolver or a different web framework).

What NOT to change

The column allow-list in compile_filters. It is the only defense against column-name injection (column names cannot be SQL-parameterized). Removing it to "accept any filter" reopens the hole.
The None-on-every-failure contract in decode_cursor. A cursor comes from the client; making it raise turns a tampered token into a 500.
One audit row per bulk batch. Emitting one per id turns a 500-id drag into 500 ledger rows; emitting zero loses the trail. The batch is the unit.
The same-404-for-absent-and-hidden rule. Distinguishing "no such id" from "hidden by your tenant scope" is a tenant-enumeration side channel.
The optimistic-rollback in the frontend factory. Dropping it leaves the UI showing changes the server rejected. If you must disable optimism for a specific entity, do it at the call site, not by removing the rollback.

FilesExpand file tree

FORKING.md

Latest commit

History