""" Regulation to Control API client. Zero dependencies beyond the standard library — no requests, no httpx — so it drops into any environment without a dependency negotiation. from regulation_control import RegulationControl client = RegulationControl() # reads REGULATION_CONTROL_API_KEY client = RegulationControl("sp_live_…") # or pass it explicitly Get a free key: curl -X POST https://controlgraph-api.com/v1/keys \ -H 'content-type: application/json' -d '{"email":"you@example.com"}' WHAT THIS IS: arithmetic and graph mapping over inputs you supply — the obligations you have read out of a regulatory change, and your own control inventory. It does not read, interpret or apply law, and it is NOT legal advice. Any penalty amount is an INTEGER number of minor units (cents). A fractional amount is rejected by the API rather than rounded. """ from __future__ import annotations import json as _json import os import urllib.error import urllib.request __all__ = [ "RegulationControl", "ApiError", "OBLIGATION_TYPES", "EVIDENCE_KINDS", "CHANGE_TYPES", "CONTROL_STATUSES", "COVERAGE_STATUSES", "FINDING_CODES", "DEADLINE_KINDS", "PRIORITY_BANDS", ] #: Override with REGULATION_CONTROL_BASE_URL, or pass base_url=. The generated #: API reference at /docs.html always carries the origin this deployment is on. DEFAULT_BASE_URL = os.environ.get("REGULATION_CONTROL_BASE_URL", "https://controlgraph-api.com") #: Branch on these rather than on the human-readable prose, which may change. #: GET /v1/obligation-types serves the same list with descriptions, the #: preparation lead time for each type, and the exact priority weights. OBLIGATION_TYPES = ( "disclosure", "recordkeeping", "reporting", "notification", "consent", "data_protection", "access_control", "monitoring", "risk_assessment", "governance", "training", "third_party_oversight", "prohibition", ) EVIDENCE_KINDS = ( "policy_document", "control_test_result", "system_configuration", "audit_log", "training_record", "attestation", "risk_assessment", "third_party_report", "notification_record", "retention_schedule", ) #: new_rule and amendment set an evidence cut-off at publishedAt: evidence #: gathered before a requirement existed cannot demonstrate it. guidance and #: enforcement_action set none. A repeal is not modelled — it creates no #: obligation; send the surviving obligations as an amendment. CHANGE_TYPES = ("new_rule", "amendment", "guidance", "enforcement_action") CONTROL_STATUSES = ("implemented", "planned", "in_remediation", "retired") #: gap = nothing matched. weak = something matched but cannot satisfy it. #: covered = an operating control plus every required evidence kind. COVERAGE_STATUSES = ("covered", "weak", "gap") FINDING_CODES = ( "no_control_mapped", "control_not_implemented", "control_retired", "control_untested", "control_stale", "control_unowned", "evidence_missing", "evidence_undated", "evidence_predates_change", "deadline_passed", "prepare_by_passed", ) DEADLINE_KINDS = ("prepare_by", "effective", "transition_end") PRIORITY_BANDS = ("critical", "high", "medium", "low") class ApiError(Exception): """ Raised for any non-2xx response. NOT raised when an obligation comes back as a ``gap`` — that is a successful answer to a legitimate question, and usually the answer you bought the API for. On a 400, ``details["path"]`` names the exact field that failed validation. """ def __init__(self, status: int, code: str, message: str, request_id: str | None = None, details=None): super().__init__(f"[{status} {code}] {message}") self.status = status self.code = code self.message = message self.request_id = request_id self.details = details class RegulationControl: def __init__(self, api_key: str | None = None, *, base_url: str = DEFAULT_BASE_URL, timeout: float = 30.0): key = api_key or os.environ.get("REGULATION_CONTROL_API_KEY") if not key: raise ValueError( "No API key. Pass one to RegulationControl(...) or set " "REGULATION_CONTROL_API_KEY. Create a free key: POST " '{}/v1/keys with {{"email": "you@example.com"}}'.format(base_url) ) self.api_key = key self.base_url = base_url.rstrip("/") self.timeout = timeout # -- transport --------------------------------------------------------- def _request(self, method: str, path: str, *, body=None, auth: bool = True) -> dict: data = _json.dumps(body).encode() if body is not None else None req = urllib.request.Request(self.base_url + path, data=data, method=method) if auth: req.add_header("Authorization", f"Bearer {self.api_key}") req.add_header("Accept", "application/json") if data: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=self.timeout) as res: return _json.loads(res.read().decode() or "{}") except urllib.error.HTTPError as e: raw = e.read().decode() try: err = _json.loads(raw).get("error", {}) except Exception: err = {} raise ApiError( e.code, err.get("code", "unknown"), err.get("message", raw[:200]), err.get("requestId"), err.get("details"), ) from None # -- API --------------------------------------------------------------- def health(self) -> dict: """Liveness and deployed version. Does not require a key.""" return self._request("GET", "/health", auth=False) def map_change(self, *, controls: list, change: dict | None = None, changes: list | None = None) -> dict: """ Map one regulatory change, or up to 50, against one control inventory. Billed one unit per CHANGE, not per control or per obligation — the inventory is shared across the batch, so re-sending 400 controls to map a second change costs nothing extra. Returns, per change: coverage for every obligation (covered / weak / gap, deny-by-default), the controls each one matched and why, the evidence now required, three dated deadlines including the derived prepare-by date, a reproducible priority derivation, and the impact graph with a reason on every edge. """ if (change is None) == (changes is None): raise ValueError("Pass exactly one of change= or changes=.") payload: dict = {"controls": controls} if changes is not None: payload["changes"] = changes else: payload["change"] = change return self._request("POST", "/v1/changes", body=payload) def demo_map(self, *, controls: list, change: dict) -> dict: """The real engine with no key: one change, 5 obligations, 20 controls.""" return self._request("POST", "/v1/demo/map", body={"controls": controls, "change": change}, auth=False) def obligation_types(self) -> dict: """ The code catalogue: every enum with its meaning, the preparation lead time per obligation type, and the exact priority weights — so a score in a response can be recomputed on paper. """ return self._request("GET", "/v1/obligation-types", auth=False) # -- convenience ------------------------------------------------------- @staticmethod def gaps(mapped: dict) -> list: """Obligations with no control mapped at all. The work nobody has started.""" return [o for o in mapped["obligations"] if o["coverage"] == "gap"] @staticmethod def weaknesses(mapped: dict) -> list: """Obligations with a control that cannot currently satisfy them.""" return [o for o in mapped["obligations"] if o["coverage"] == "weak"] @staticmethod def due_within(mapped: dict, days: int) -> list: """Obligations whose compliance date is inside ``days`` — negative once passed.""" return [o for o in mapped["obligations"] if o["daysToCompliance"] <= days] @staticmethod def create_key(email: str, *, base_url: str = DEFAULT_BASE_URL, name: str | None = None) -> dict: """Create a free sandbox key. The key is returned once and never again.""" payload: dict = {"email": email} if name: payload["name"] = name req = urllib.request.Request( base_url.rstrip("/") + "/v1/keys", data=_json.dumps(payload).encode(), method="POST" ) req.add_header("Content-Type", "application/json") with urllib.request.urlopen(req, timeout=30) as res: return _json.loads(res.read().decode())