"""The anti-corruption layer: the three neighbours a lead is converted into, in the lead's language.
Every call to another brick goes through here, and nowhere else. Each neighbour speaks its own model —
`bp_key` and `legal_name` (bp), `full_name` (activity), `source_ref` and stage names (pipeline) — and
answers in its own shapes and status codes. This module translates both ways, so the use cases say
"make this lead's company a customer" and get back a bp_id or a `NeighbourFailed`, and a neighbour that
renames a field is fixed in one place.
The seams (erp_platform.seam) and the permission the lead brick's service identity needs on each:
PARTNERS business-partner bp:partner:promote POST /partners/promote, GET /partners?bp_key=
CONTACTS activity activity:contact:write POST /contacts
DEALS pipeline pipeline:opportunity:link_source POST /opportunities/from-source (pipeline 0.3.0: a
machine's route — no person can name a source_ref, so no rep can plant the deal another's
conversion lands on; a ref whose deal is another partner's is refused 409)
Not installed = configured nowhere (`installed` is False). Down = installed, but the call came back without
an answer the lead can use: that is a `NeighbourFailed` with `answered=False`; a refusal in the
neighbour's own words is `answered=True`.
"""
import json
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam
from brick_lead.domain.lead import Lead
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=["bp:partner:promote"])
CONTACTS = Seam("activity", "ACTIVITY_BASE_URL", roles=["service"], permissions=["activity:contact:write"])
DEALS = Seam("pipeline", "PIPELINE_BASE_URL", roles=["service"], permissions=["pipeline:opportunity:link_source"])
CONVERSION = (PARTNERS, CONTACTS, DEALS)
WHAT = {"business-partner": "the partner register", "activity": "the contact list", "pipeline": "the pipeline"}
class NeighbourFailed(Exception):
"""A neighbour did not do what was asked. `answered`: it refused (its reason is `said`), as opposed to
not answering at all (down, unreachable, a 5xx)."""
def __init__(self, seam: Seam, reply):
self.what, self.status = WHAT.get(seam.name, seam.name), reply.status
self.answered = bool(reply.status) and 400 <= reply.status < 500
try:
self.said = json.loads(reply.reason).get("detail") or reply.reason
except (ValueError, AttributeError):
self.said = reply.reason
super().__init__(f"{self.what}: {self.said}")
def missing_for_conversion() -> list:
"""The neighbours conversion needs that are not installed."""
return [s.name for s in CONVERSION if not s.installed]
def make_customer(lead: Lead, actor: str) -> int:
"""The lead's company as a business partner with the customer role — created, or reused when its domain
already is one (BP-04 promote, idempotent by bp_key). Returns the bp_id (BIGINT)."""
r = PARTNERS.post("/partners/promote", {"bp_key": lead.domain, "role": "customer",
"legal_name": lead.company or None, "country": lead.country or None,
"domain": lead.domain}, actor)
if not r:
raise NeighbourFailed(PARTNERS, r)
return int(r.data["bp_id"])
def add_contact(lead: Lead, bp_id: int, actor: str) -> str:
"""The lead's person as a contact at the partner — reused when the email already is one there."""
r = CONTACTS.post("/contacts", {"bp_id": int(bp_id), "full_name": lead.name, "email": lead.email,
"title": lead.title}, actor)
if not r:
raise NeighbourFailed(CONTACTS, r)
return str(r.data["contact_id"])
def open_deal(lead: Lead, bp_id: int, actor: str, *, currency: str, name: Optional[str],
amount: Optional[Decimal], close_date: Optional[str]) -> str:
"""A deal for the partner in the pipeline's first stage, named for the lead's company unless named, owned by
whoever converts (`actor`). `source_ref` "lead:<lead_id>" makes it ONE deal per lead, however often this is
retried; a ref that already opened a deal for ANOTHER partner is refused by the pipeline (409)."""
r = DEALS.post("/opportunities/from-source",
{"source_ref": f"lead:{lead.lead_id}", "bp_id": str(int(bp_id)), "owner": actor,
"name": (name or "").strip() or lead.company or lead.domain, "currency": currency,
"amount": None if amount is None else str(amount), "close_date": close_date}, actor)
if not r:
raise NeighbourFailed(DEALS, r)
return str(r.data["opportunity_id"])
def is_customer(domain: str, actor: str) -> Optional[bool]:
"""Whether the partner register holds this domain as a customer (its bp_key). None when it cannot say:
not installed, or not answering."""
if not PARTNERS.installed:
return None
r = PARTNERS.get("/partners", params={"bp_key": domain}, actor=actor)
if not r.ok:
raise NeighbourFailed(PARTNERS, r)
return any("customer" in p.get("roles", []) for p in r.data.get("partners", []))