Autoschematic

Connector.plan() is where a connector compares "current state" and "desired state" and produces a series of connector ops. Each op is returned in a PlanResponseElement, and the ordered list of ops forms the plan.

@dataclass
class PlanResponseElement:
    op_definition: str
    writes_outputs: list[str] = field(default_factory=list)
    friendly_message: str = ""

The Grafana connector uses small JSON blobs as its op format. The address tells op_exec() which remote object to mutate, so the op itself only needs to describe the action and, when relevant, the desired body.

match [current, desired]:
    case [_, None]:
        return [
            PlanResponseElement(
                op_definition=json.dumps({"action": "delete", "type": resource_type}),
                writes_outputs=[],
                friendly_message=f"Delete {resource_type}",
            )
        ]
    case [None, desired]:
        return [
            PlanResponseElement(
                op_definition=json.dumps({
                    "action": "create",
                    "body": json.loads(desired),
                }),
                writes_outputs=["id"],
                friendly_message=f"Create {resource_type}",
            )
        ]
    case [current, desired] if current != desired:
        return [
            PlanResponseElement(
                op_definition=json.dumps({
                    "action": "update",
                    "body": json.loads(desired),
                }),
                writes_outputs=["id"],
                friendly_message=f"Update {resource_type}",
            )
        ]

op_exec() parses the op string and dispatches by address:

async def op_exec(self, addr: str, op: str) -> OpExecResponse:
    dec_op = json.loads(op)
    action = dec_op["action"]

    match decode_addr(addr):
        case DatasourceAddress(name):
            return await self.exec_datasource(name, action, dec_op.get("body"))
        case DashboardAddress(uid):
            return await self.exec_dashboard(uid, action, dec_op.get("body"))
        case _:
            raise InvalidAddr(addr)

That separation is important. plan() should stay about diffing and sequencing, while op_exec() deals with API-specific mechanics. In Grafana, datasource updates need an extra read to discover the numeric datasource ID before issuing the PUT, so that lookup lives in exec_datasource() rather than in plan().

Tips for connector ops:

• Do not duplicate path data inside the op if addr already identifies the resource.
• Use readable friendly_message values so autoschematic plan output makes sense to the user.

Reference files:

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

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

Next: Outputs and Virtual Addresses