Writing a provider

A provider is the only component that knows where a resource's data actually lives. Implementing one means answering the query language; everything above — fields, views, routes, templates — comes for free.

The interface

class Provider(Protocol):
    name: str
    capabilities: Capabilities

    async def list(self, q: ListQuery, ctx: Ctx) -> Page[Record]: ...
    async def get(self, pk: Any, ctx: Ctx) -> Record | None: ...
    async def create(self, data: dict, ctx: Ctx) -> WriteResult: ...
    async def update(self, pk: Any, data: dict, ctx: Ctx) -> WriteResult: ...
    async def delete(self, pk: Any, ctx: Ctx) -> WriteResult: ...
    async def aggregate(self, spec: AggSpec, ctx: Ctx) -> list[dict]: ...
    async def update_if(
        self, pk: Any, data: dict, expect: dict, ctx: Ctx
    ) -> WriteResult | None: ...
    async def warm(self) -> None: ...
    async def health(self) -> tuple[bool, str]: ...

Subclass BaseProvider and implement only what your backend supports; the rest refuse politely.

Two of those are worth explaining, because their defaults are deliberate.

warm() — pay the startup cost at startup

Whatever is the same for every request and expensive the first time: reflecting a table, resolving a schema, opening a channel. Left until the first request, it lands on whoever happens to arrive first and shows up as one slow page after every deploy.

async def warm(self) -> None:
    await self.table()          # reflect now, not on the first request

The registry calls this for every resource concurrently once providers are bound, and tolerates failure: a backend that is temporarily unreachable delays its own resource, it does not stop the application starting. It is logged, though — a warning here is usually the first sign of a misspelled table name, and finding that at startup beats finding it as a 502 an hour later.

update_if() — a write that loses the race, safely

Update only while the stored record still matches expect; return None when it does not, because somebody else got there first.

async def update_if(self, pk, data, expect, ctx):
    stmt = update(table).where(table.c.id == pk, table.c.sent_at.is_(None))
    result = await conn.execute(stmt.values(**data))
    return None if result.rowcount == 0 else WriteResult.success(...)

This is what makes background work safe to run from more than one place. The sweep behind crm notify-due claims each row by writing the very column its query filters on, so the winner's write removes the row from every other sweeper's result set and a reminder cannot be delivered twice. The same primitive is what an edit form needs to say "this record changed while you were looking at it" instead of silently overwriting a colleague.

Do not emulate it with a read and a write. The default raises UnsupportedOperation, and that is the right answer for a backend that cannot express a conditional write: read-then-write passes every test and still loses rows in production, which is worse than an honest refusal. The shim does not emulate it either, for the same reason.

Declare capabilities honestly

This is the one rule that matters.

self.capabilities = Capabilities(
    read=True,
    server_filter=True,
    filter_ops=frozenset({Op.EQ, Op.IN}),  # only what you really support
    server_sort=False,                     # the shim will sort
)

Declaring False is not a failure — it is a request for the shim to handle that concern, and it will. Declaring True is a promise, because the shim then steps aside entirely. A provider claiming server_filter=True while ignoring an operator returns silently wrong results that nothing will catch.

If in doubt, declare less. The cost is a wider read; the cost of over-claiming is incorrect data on screen.

The contract suite

# tests/contract/test_provider_contract.py
CAPABILITY_PROFILES = {
    "native": ...,           # your provider as it is
    "shimmed-dumb": ...,     # with every capability stripped
    "shimmed-partial": ...,  # with some stripped
}

Add your provider to this suite. It runs the same ~90 assertions against each profile and requires identical results. If your provider passes natively but fails when shimmed, its capability declaration is lying.

Registering it

Two decorators wire a new backend in:

@register_connection("mybackend", close=..., check=...)
async def open_mybackend(spec: ConnectionSpec):
    return MyClient(spec.option("url", required=True))

@register_provider_factory("mybackend")
def build(handle, target: str, resource) -> MyProvider:
    return MyProvider(handle, target, pk_field=resource.pk)

Then in connections.yaml:

connections:
  my.backend:
    type: mybackend
    url: ${MY_BACKEND_URL}

And on a resource: provider="my.backend#some_target". Whatever follows the # reaches your factory as target — a table name, an endpoint path, a routing key.

Shipped providers

Provider	Notes
SQLProvider	Any SQLAlchemy-supported database. Tables are reflected, so no model layer is needed. Fully capable.
RestProvider	Declarative mapping of an HTTP API: where rows live, how pages work, which filters push down.
AMQPWriteProvider	Publishes commands. Writes return PENDING; reads are refused rather than faked.
MemoryProvider	Tests, demos, prototypes. Capabilities configurable, which is how the shim is tested.
CompositeProvider	Reads from one, writes to another. Declare with write_provider= on the resource: a connection name, a provider, or a factory called with (registry, resource).
UnionProvider	Reads one resource from several backends at once. Declare with provider=Union(...); see Several databases.
ReadOnly	Wraps any provider to refuse writes.
AuditingProvider	Wraps any provider to record every write, with before/after values. Applied automatically; not something a resource declares.

A table the database does not have

A resource names db.main#thing, and thing is normally a table. It may instead be a derived target: a SELECT declared in the module that needs it, aliased under a name and used everywhere a table would be.

from app.providers.sql import register_selectable

def occupancy(tables):
    venues, bookings = tables["venues"], tables["bookings"]
    return (
        select(venues.c.id.label("id"), venues.c.name.label("name"),
               func.coalesce(func.sum(bookings.c.guests), 0).label("guests"))
        .select_from(venues.outerjoin(bookings, bookings.c.venue_id == venues.c.id))
        .group_by(venues.c.id, venues.c.name)
    )

register_selectable("venue_occupancy", tables=("venues", "bookings"),
                    build=occupancy, description="Guests booked per venue.")

Resource("occupancy", provider="db.main#venue_occupancy", pk="id", fields=[...])

The reason this exists rather than "create a view": a database read through a provider is frequently owned by whatever service migrates it, so this application may read it and may not add anything to it. A derived target keeps the definition here, versioned with the screen it feeds.

Everything downstream — filters, sorting, grouping, the shim, every view — sees a table like any other, because the statement is aliased as a subquery before the provider composes onto it. Its sources are reflected through the same cache the tables use, so it costs no extra catalogue queries.

Two consequences worth stating:

• It is read-only by construction. There is no row behind a computed one, so the factory builds a read-only provider and a write is refused rather than attempted. Send the write somewhere that will accept it — see write_provider in the table above.
• The resource must name its key. A subquery declares no primary key, so pk= is the only thing that can address a row, and the SELECT has to include it.

REST connection authentication

connections:
  crm_api:
    type: rest
    options:
      base_url: https://api.example.com
      auth:
        type: oauth2                       # bearer | header | basic | oauth2
        token_url: https://issuer/oauth/token
        client_id: ${API_CLIENT_ID}
        client_secret: ${API_CLIENT_SECRET}
        scope: read:things                 # optional
        client_auth: basic                 # basic (default) | body
        extra: {audience: https://api.example.com}   # Auth0 wants this
        leeway: 30                         # renew this many seconds early

bearer, header and basic are a header computed once, when the connection opens. oauth2 is client-credentials, and is the only one with a lifecycle: the token is fetched at first use, reused until shortly before it expires, and renewed once if the API rejects it with a 401. Concurrent requests against a cold connection share one token fetch rather than each opening their own.

An unrecognised type is refused when the connection opens. Ignoring it would build an unauthenticated client, and the resulting 401 from the API reads like wrong credentials rather than a misspelled type name.

Acting as the signed-in user

Every mode above authenticates as the application, so the API sees one identity no matter who clicked. When the API is the real system of record that is usually wrong: its audit trail should name the person.

      auth:
        type: caller_token
        fallback:                          # used when the caller has no token
          type: oauth2
          token_url: https://issuer/oauth/token
          client_id: ${API_CLIENT_ID}
          client_secret: ${API_CLIENT_SECRET}

The header is built per request from the caller's own OIDC access token, which means two things must both be true: the deployment sets CRM_OIDC_KEEP_ACCESS_TOKEN=true, and the caller signed in through OIDC.

fallback covers everyone else — a background job, the CLI, a caller who signed in with a password — and is an ordinary auth block of any of the types above. Without it those callers make unauthenticated requests, which is occasionally what you want and usually a 401.

Two costs, stated plainly rather than discovered later:

• The session cookie is signed, not encrypted. Anyone who can read the cookie can read the token in it. It is HttpOnly, Secure and SameSite, and the holder is the person whose token it is — but this is a decision to take deliberately, which is why the setting is off by default.
• A token is large. If the cookie will not fit in 4 KB the session is stored without the token and a warning is logged: writes then fall back to the connection's credentials. The alternative is a cookie the browser silently discards, which signs the person out on their next click.

Only the access token is kept. A refresh token would let whoever reads the cookie mint new tokens long after the session ended — a much larger promise than "act as this person while they are signed in".

Asynchronous writes

If your backend accepts a write without confirming it, say so:

self.capabilities = Capabilities(write=True, write_mode="async")

async def create(self, data, ctx):
    await self.publish(data)
    return WriteResult.pending(message="Queued for processing.")

The form pipeline will render a queued badge with the correlation reference rather than claiming the record was saved. Nothing else needs to change.