dodil/erp-brick-hr

src/brick_hr/infrastructure/contracts_acl.py

"""Anti-corruption layer: contracts (context map: customer/supplier — hr is the customer).

hr asks through the contracts.* CAPABILITIES (capabilities.yaml 0.4), never reads contracts' tables, and contracts
never reads hr's: hr passes the counterparty bp_id, the context ref (person_id|n), the contract type and the
VARIABLES (never clause text), including the UK client_size_regime from hr's settings.

  draft   idempotency_key = '<source_ref>|<type_code>' (contract_id is derived from it, so a retried step drafts
          one contract); source_ref = 'hr:<run_id>#<step_no>' — the 'hr:' namespace is what makes hr the
          ORIGINATOR, the only caller that is given the executed copy;
  send    freezes the draft and sends it for signature;
  status  polled each tick until signed, declined or expired;
  fetch_executed  the signed copy as base64 with its sha256 (never an object URL), filed on the person as a
          pay-bearing document.

With contracts not installed every call is a ManualTask carrying the same input ("send the employment contract
and attach the signed copy").
"""
from typing import Optional

from brick_hr.domain.run import Outcome
from brick_hr.infrastructure import capability_acl as acl


def source_ref(run_id: str, step_no: str) -> str:
    return f"hr:{run_id}#{step_no}"


def draft_input(type_code: str, counterparty_bp_id: int, context_ref: str, src: str, variables: dict,
                signers: Optional[list]) -> dict:
    return {"type_code": type_code, "counterparty_bp_id": int(counterparty_bp_id), "context_ref": context_ref,
            "source_ref": src, "variables": variables, "signers": signers}


def draft(data: dict, *, dry_run: bool = False, actor: str = "") -> Outcome:
    return acl.call("contracts.draft", f"{data['source_ref']}|{data['type_code']}", data, dry_run=dry_run, actor=actor)


def send(contract_id: str, key: str, actor: str = "") -> Outcome:
    return acl.call("contracts.send", key, {"contract_id": contract_id}, actor=actor)


def status(contract_id: str, actor: str = "") -> Outcome:
    return acl.call("contracts.status", f"status|{contract_id}", {"contract_id": contract_id}, actor=actor)


def fetch_executed(contract_id: str, actor: str = "") -> Outcome:
    return acl.call("contracts.fetch_executed", f"fetch|{contract_id}", {"contract_id": contract_id}, actor=actor)


__all__ = ["source_ref", "draft_input", "draft", "send", "status", "fetch_executed"]