Connect planning to your system of record

# plans_service.py
# A complete plan provider. It answers every method the plan data plane sends,
# and it passes the shared store conformance suite as written.

import json
import sys

PLANS = {}


def send(message):
    sys.stdout.write(json.dumps(message) + "\n")
    sys.stdout.flush()


def result(request_id, value):
    send({"jsonrpc": "2.0", "id": request_id, "result": value})


def failure(request_id, code, message, extra=None):
    data = {"code": code}
    if extra:
        data.update(extra)
    send(
        {
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": -32001, "message": message, "data": data},
        }
    )


def plans_for(owner):
    return PLANS.setdefault(owner, {})


def handle(request):
    request_id = request["id"]
    method = request["method"]
    params = request.get("params") or {}
    owner = params.get("owner", "")

    if method == "initialize":
        result(request_id, {"protocol_version": 1, "provider_kind": "acme-tracker"})
        return
    if method == "shutdown":
        result(request_id, None)
        return

    store = plans_for(owner)

    if method == "plans/create":
        document = params["document"]
        store[document["id"]] = document
        result(request_id, document)
        return

    if method == "plans/read":
        document = store.get(params["id"])
        if document is None:
            failure(request_id, "plan_not_found", "Plan not found: %s" % params["id"])
            return
        result(request_id, document)
        return

    if method == "plans/list":
        query = params.get("input") or {}
        rows = list(store.values())
        if query.get("status") is not None:
            rows = [row for row in rows if row["status"] == query["status"]]
        if query.get("retention") is not None:
            rows = [row for row in rows if row["retention"] == query["retention"]]
        rows.sort(key=lambda row: row["created_at"], reverse=True)
        offset = int(query.get("cursor") or 0)
        limit = max(1, min(100, int(query.get("limit") or 20)))
        page = rows[offset : offset + limit]
        payload = {"plans": page}
        if offset + len(page) < len(rows):
            payload["next_cursor"] = str(offset + len(page))
        result(request_id, payload)
        return

    if method == "plans/write":
        current = store.get(params["id"])
        if current is None:
            failure(request_id, "plan_not_found", "Plan not found: %s" % params["id"])
            return
        expected = params["expected"]
        if (
            current["revision"] != expected["revision"]
            or current["digest"] != expected["digest"]
            or current["spec_digest"] != expected["specDigest"]
        ):
            failure(
                request_id,
                "plan_conflict",
                "Plan changed since it was read",
                {"reason": "cas"},
            )
            return
        store[params["id"]] = params["document"]
        result(request_id, params["document"])
        return

    if method == "plans/reconcile":
        document = store.get(params["id"])
        if document is None:
            failure(request_id, "plan_not_found", "Plan not found: %s" % params["id"])
            return
        result(request_id, document)
        return

    if method == "plans/delete":
        result(request_id, store.pop(params["id"], None) is not None)
        return

    failure(request_id, "plan_invalid", "unknown method %s" % method)


for line in sys.stdin:
    line = line.strip()
    if line:
        handle(json.loads(line))