Autoschematic

To read remote resources, a connector provides two methods:

• list() enumerates which resource addresses currently exist remotely.
• get() fetches the current state of one address and returns it as bytes plus optional outputs.

The Grafana connector's list() method does not try to be clever. It asks the API for all datasources and dashboards, then turns those results into addresses:

(Note: this guide omits talking about subpaths(), which is how a connnector can split up its list() calls so they're not so gigantic. TODO!)

async def list(self, subpath: str) -> list[str]:
    addrs: list[str] = []

    status, datasources = await self._api_get("/api/datasources")
    if status == 200 and isinstance(datasources, list):
        for ds in datasources:
            name = ds.get("name", "")
            if name:
                addrs.append(f"grafana/datasources/{name}.json")

    status, results = await self._api_get("/api/search?type=dash-db")
    if status == 200 and isinstance(results, list):
        for db in results:
            uid = db.get("uid", "")
            if uid:
                addrs.append(f"grafana/dashboards/{uid}.json")

    return addrs

get() then dispatches to resource-specific helpers:

async def get(self, addr: str) -> GetResponse:
    match decode_addr(addr):
        case DashboardAddress(uid):
            return await self._get_dashboard(uid)
        case DatasourceAddress(name):
            return await self._get_datasource(name)
        case _:
            return GetResponse(exists=False)

async def _get_datasource(self, name: str) -> GetResponse:
    status, body = await self._api_get(f"/api/datasources/name/{name}")

    match status:
        case 404:
            return GetResponse(exists=False)
        case 200:
            assert isinstance(body, dict)

            ds_id = str(body.get("id", ""))
            for field in DATASOURCE_TRANSIENT_FIELDS:
                body.pop(field, None)

            return GetResponse(
                exists=True,
                resource_definition=json.dumps(body, indent=2).encode(),
                outputs={"id": ds_id},
            )
            
        case _:
            raise RuntimeError(f"Failed to get datasource '{name}': {status} {body}")
            

async def _get_dashboard(self, uid: str) -> GetResponse:
    [...]

If a resource doesn't exist, get() should not throw an error, but simply return GetResponse(exists=false).

Important behavior to preserve:

• resource_definition must be bytes. The Grafana example uses json.dumps(body, indent=2).encode().
• Missing resources should return GetResponse(exists=False).
• list() returns paths only. It should not fetch every full body unless the API gives you no better option.
• Keep your read methods deterministic. If an API returns unstable ordering, sort the final address list before returning it, and likewise for fields in a struct returned by get().

Reference files:

• autoschematic-sdk-python/examples/grafana.py

• autoschematic-sdk-python/autoschematic_sdk/types.py

Next: Planning and Executing Connector Ops