"""The use cases — what reps, managers, other bricks and their agents can do with the pipeline.
Each is declared once and served as an API route AND an MCP tool.
WHO (CRM spec: "a rep sees rows they own; a manager sees the team's"; CRM-01): a rep sees and changes
their own deals; a manager or analyst sees every deal, and a manager may change any. A deal the caller
may not see is 404, never 403. From 0.4.0 "manager" is a set of PERMISSIONS, not a role name: seeing every deal is
pipeline:opportunity:read_all, changing anyone's pipeline:opportunity:manage (roles.yaml: the default roles).
NEIGHBOURS, asked through their routes (erp_platform.seam), never their tables:
quote "which quotes on this deal are accepted — for which customer, in which currency, for how much?" —
the evidence won needs (0.3.0: only an acceptance by THIS deal's customer, in its currency, and not
voided by a reopen, counts; the deal then takes that total).
Not installed: won is recorded without it, and the history says so. Installed but down:
won is refused for now — unlike a sale, marking a deal won can wait for a retry, and a won
deal recorded without its evidence would be the number leadership is given.
business-partner "who is this customer, and who is in their corporate family?" — CRM-08.
"""
from contextlib import contextmanager
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Literal, Optional
from uuid import uuid4
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from erp_platform.money import CurrencyMismatch, InvalidMoney, Money
from erp_platform.repository import Conflict, retry_on_conflict
from erp_platform.seam import Seam
from erp_platform.types import Amount, BigId, Email, IsoDate, IsoTime, Name, Note, Percent, text
from brick_pipeline.domain.pipeline import (DEFAULT_PIPELINE, EvidenceMissing, Opportunity, Pipeline,
PipelineRuleBroken, RiskPolicy, Stage, forecast, period_of)
from brick_pipeline.infrastructure.tables import SourceRefTaken
WRITE, TOUCH = "pipeline:opportunity:write", "pipeline:opportunity:touch"
OVERRIDE, SNAPSHOT, STAGES = "pipeline:forecast:override", "pipeline:forecast:snapshot", "pipeline:stages:write"
# 0.3.0. REOPEN: a manager's. LINK_SOURCE: a MACHINE's — held by no person role, granted to the lead brick's seam
# (tier 1) and its service client (tier 2): naming the source a deal is opened from is how a conversion finds its
# deal again, so a person who could name one could plant a deal for someone else's lead (the CRM audit, C2).
REOPEN, LINK_SOURCE = "pipeline:opportunity:reopen", "pipeline:opportunity:link_source"
# 0.4.0: a brick decides by PERMISSION, never by role name (erp-platform 0.9.2). Seeing every owner's deals was the
# `manager`/`analyst` ROLES, changing them the `manager` role; they are now these (roles.yaml: the default roles).
READ_ALL, MANAGE = "pipeline:opportunity:read_all", "pipeline:opportunity:manage"
Category = Literal["pipeline", "best_case", "commit", "omitted"]
Currency, StageName, Ref, Reason = text(8), text(64), text(200), text(64)
# 0.3.1: LINK_SOURCE is a MACHINE permission (erp-platform 0.9.2) — passed on an exact grant only, so `*:*:*` never
# reaches it; and the process is declared (process.yaml → GET /process/bpmn).
PIPE = Brick("pipeline", version="0.4.1", ids_as_strings=True, id_keys=("group",), machine_permissions=(LINK_SOURCE,),
instructions=(
"The sales pipeline: deals (opportunities) for business partners, moving through the business's own "
"stages. Each stage sets a deal's probability and forecast category. A deal is won when the customer "
"has accepted a quote; it is lost with a reason from the agreed list. The forecast is per rep, quarter "
"and currency: open, weighted (amount × probability) and committed. Money is a string, two decimals. "
"Won takes the accepted quote's total; a closed deal is final until a manager reopens it with a reason. "
"bp_id is a 64-bit id, sent and returned as digits. A deal carries a `version`: pass it back with a change, "
"and a 409 means someone changed it since."))
PIPE.declare_process("process.yaml")
QUOTES = Seam("quote", "QUOTE_BASE_URL", roles=["service"], permissions=["quote:opportunity:read"])
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])
@contextmanager
def _repository_default():
from erp_platform import db
from brick_pipeline.infrastructure.tables import PipelineRepository
with db.engine.begin() as conn:
yield PipelineRepository(conn)
repository = _repository_default
def _aware(iso: str) -> datetime:
"""An ISO time as an aware datetime; one without a zone is taken as UTC."""
try:
d = datetime.fromisoformat(str(iso).replace("Z", "+00:00"))
except ValueError:
raise Refused(422, f"{iso!r} is not a time (ISO 8601)") from None
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
def _today() -> str:
return date.today().isoformat()
@contextmanager
def _rules():
try:
yield
except EvidenceMissing as e:
raise Refused(422, str(e)) from None
except (PipelineRuleBroken, CurrencyMismatch) as e:
raise Refused(409, str(e)) from None
except InvalidMoney as e:
raise Refused(422, str(e)) from None
def _sees(caller: dict, o: Opportunity) -> bool:
return auth.owner_scope(caller, see_all=READ_ALL) is None or o.owner == caller.get("email")
def _load(repo, caller: dict, opportunity_id: str, *, write: bool = False) -> Opportunity:
o = repo.get(opportunity_id)
if o is None or not _sees(caller, o):
raise Refused(404, f"no opportunity {opportunity_id}")
if write and not (o.owner == caller.get("email") or auth.has_permission(caller, MANAGE)):
raise Refused(403, f"opportunity {opportunity_id} belongs to {o.owner}; only they or a manager "
f"can change it")
return o
def _at(o: Opportunity, version: Optional[int]) -> Opportunity:
"""The deal, if it is still at the version the caller loaded; a Conflict (409) if not."""
if version is not None and o.version != int(version):
raise Conflict("pipeline_opportunities", {"opportunity_id": o.opportunity_id}, int(version), o.version)
return o
def _changing(version: Optional[int], fn):
"""With the caller's version, a race is their 409; with none, the change is re-run on the latest deal —
two people editing one deal at once both land (the audit lost 6 of 6)."""
return fn() if version is not None else retry_on_conflict(fn)
def _view(o: Opportunity, p: Pipeline, risk: RiskPolicy) -> dict:
w = o.weighted(p)
return {"opportunity_id": o.opportunity_id, "bp_id": o.bp_id, "name": o.name, "owner": o.owner,
"stage": o.stage, "is_open": o.is_open(p), "currency": o.currency,
"amount": None if o.amount is None else o.amount.amount,
"probability": o.probability_for(p), "probability_overridden": o.probability is not None,
"weighted": None if w is None else w.amount, "category": o.category(p),
"category_reason": o.category_reason, "close_date": o.close_date,
"period": period_of(o.close_date), "loss_reason": o.loss_reason,
"last_activity_at": o.last_activity_at, "at_risk": o.risk(p, risk),
"won_quote_ids": list(o.won_quote_ids), "version": o.version}
def _changed(repo, caller: dict, o: Opportunity, action: str, detail: str = "") -> dict:
o.last_activity_at = datetime.now(timezone.utc).isoformat() # a change is activity
repo.save(o)
repo.record(o.opportunity_id, action, caller["email"], detail)
return _view(o, *repo.pipeline())
# ── reading ────────────────────────────────────────────────────────────────────────────────
@PIPE.use_case(kind="query", route="GET /opportunities")
def list_opportunities(caller: dict, mine: bool = False, open_only: bool = True,
at_risk: bool = False) -> dict:
"""The deals the caller may see (CRM-01: a rep sees their own; a manager sees everyone's). `at_risk`
narrows to late-stage deals that have gone quiet or passed their close date (CRM-16)."""
with repository() as repo:
p, risk = repo.pipeline()
out = [_view(o, p, risk) for o in repo.deals(owner=auth.owner_scope(caller, mine, see_all=READ_ALL))]
out = [v for v in out if (v["is_open"] or not open_only) and (v["at_risk"] or not at_risk)]
return {"opportunities": out}
@PIPE.use_case(kind="query", route="GET /opportunities/{opportunity_id}", etag="version")
def get_opportunity(caller: dict, opportunity_id: Ref) -> dict:
"""One deal: its stage, probability, weighted value, forecast category, whether it is at risk, and its
`version` (send it back with a change)."""
with repository() as repo:
return _view(_load(repo, caller, opportunity_id), *repo.pipeline())
@PIPE.use_case(kind="query", route="GET /opportunities/{opportunity_id}/history")
def opportunity_history(caller: dict, opportunity_id: Ref) -> dict:
"""Every stage move, amount change and override on a deal, with who and when."""
with repository() as repo:
_load(repo, caller, opportunity_id)
return {"opportunity_id": opportunity_id, "events": repo.history(opportunity_id)}
@PIPE.use_case(kind="query", route="GET /partners/{bp_id}/group-pipeline")
def group_pipeline(caller: dict, bp_id: BigId) -> dict:
"""CRM-08: every open deal across the customer's corporate group (the partner, its parents' group and
every subsidiary), so a discount given to one subsidiary is not quoted back to another. Needs the
partner register; without it, only this partner's own deals are shown."""
members, note = {int(bp_id)}, ""
if PARTNERS.installed:
up = PARTNERS.get(f"/partners/{int(bp_id)}/family", actor=caller.get("email", ""))
if up.ok:
root = (up.data.get("parents") or [{"bp_id": int(bp_id)}])[0]["bp_id"]
fam = PARTNERS.get(f"/partners/{int(root)}/family", actor=caller.get("email", "")) \
if root != int(bp_id) else up
members |= {int(root)} | {int(m["bp_id"]) for m in fam.data.get("below", [])}
else:
note = f"group not resolved: {up.reason}"
else:
note = "no partner register installed: this partner only"
with repository() as repo:
p, risk = repo.pipeline()
deals = [o for o in repo.deals(bp_ids=sorted(members)) if _sees(caller, o) and o.is_open(p)]
return {"bp_id": bp_id, "group": sorted(members), "note": note,
"opportunities": [_view(o, p, risk) for o in deals]}
@PIPE.use_case(kind="query", route="GET /forecast")
def get_forecast(caller: dict, period: Optional[text(16)] = None, mine: bool = False) -> dict:
"""CRM-09: open, weighted and committed amounts per rep, quarter and currency — computed now from the
deals the caller may see. Deals with no amount or close date are counted as unforecastable."""
with repository() as repo:
p, _ = repo.pipeline()
rows = forecast(repo.deals(owner=auth.owner_scope(caller, mine, see_all=READ_ALL)), p)
return {"rows": [r for r in rows if period is None or r["period"] == period]}
@PIPE.use_case(kind="query", route="GET /forecast/snapshots/{snapshot_date}")
def get_forecast_snapshot(caller: dict, snapshot_date: IsoDate) -> dict:
"""CRM-15: the forecast as it stood on a date — "what did we think on the 1st?" A rep sees their own
rows; a manager sees all."""
with repository() as repo:
snap = repo.snapshot(snapshot_date)
if snap is None:
raise Refused(404, f"no forecast snapshot for {snapshot_date}")
scope = auth.owner_scope(caller, see_all=READ_ALL)
snap["rows"] = [r for r in snap["rows"] if scope is None or r["owner"] == scope]
return snap
@PIPE.use_case(kind="query", route="GET /pipeline")
def get_pipeline(caller: dict) -> dict:
"""The stages of this business's sales process, in order, with their probabilities and forecast
categories; the agreed loss reasons; and when a deal counts as at risk."""
with repository() as repo:
p, risk = repo.pipeline()
return {"stages": [{"name": s.name, "position": s.position, "probability": s.probability,
"category": s.category, "is_won": s.is_won, "is_closed": s.is_closed}
for s in p.stages],
"loss_reasons": list(p.loss_reasons),
"risk": {"quiet_days": risk.quiet_days, "from_probability": risk.from_probability}}
# ── deals ──────────────────────────────────────────────────────────────────────────────────
def _customer_may_transact(caller: dict, bp_id: int) -> None:
if PARTNERS.installed:
r = PARTNERS.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""))
if r.status == 404:
raise Refused(422, f"there is no business partner {bp_id} — add the customer first")
if r.ok and not r.data.get("can_transact", False):
raise Refused(409, f"{r.data.get('legal_name', bp_id)} is {r.data.get('status')} in the "
f"partner register, so no new deal can be opened")
def _open(caller: dict, owner: str, bp_id: int, name: str, currency: str, amount, close_date,
ref: Optional[str]) -> dict:
_customer_may_transact(caller, bp_id)
with repository() as repo, _rules():
p, _ = repo.pipeline()
o = Opportunity.open(f"O-{uuid4().hex[:10].upper()}", bp_id, name, owner, currency, p,
amount=amount, close_date=close_date)
if ref:
repo.claim_source(ref, o.opportunity_id) # first: a loser fails before writing the deal
return _changed(repo, caller, o, "opened", f"{o.name} for partner {o.bp_id} in {o.stage}"
+ (f" (from {ref})" if ref else ""))
@PIPE.use_case(kind="command", route="POST /opportunities", permission=WRITE)
def open_opportunity(caller: dict, bp_id: BigId, name: Name, currency: Currency, amount: Optional[Amount] = None,
close_date: Optional[IsoDate] = None) -> dict:
"""Open a deal for a customer (a business partner), owned by the caller, in the first open stage. Every
call opens a new deal — send an Idempotency-Key to make a retry return the first one."""
return _open(caller, caller["email"], bp_id, name, currency, amount, close_date, None)
@PIPE.use_case(kind="command", route="POST /opportunities/from-source", permission=LINK_SOURCE, idempotent=True,
tool=False)
def open_opportunity_from_source(caller: dict, source_ref: Ref, bp_id: BigId, name: Name, currency: Currency,
owner: Email, amount: Optional[Amount] = None,
close_date: Optional[IsoDate] = None) -> dict:
"""For another BRICK, not a person: open the deal for a source document — "lead:<lead_id>" when a lead is
converted (CRM-03) — owned by `owner`, the person converting. The same source_ref always returns the same
deal, so a retry whose first reply was lost gets the deal it already opened. If the ref already opened a
deal for ANOTHER partner, it is refused 409 and nothing is written: the conversion stops there.
The ref is claimed with a plain INSERT into pipeline_source_refs (a duplicate fails 23505 → the winner is
read), never an upsert. The permission is held by no person role (C2)."""
ref = source_ref.strip()
if not ref:
raise Refused(422, "source_ref names the document the deal is opened from")
found = _opened_from(ref, bp_id)
if found is not None:
return found
try:
return _open(caller, owner.strip().lower(), bp_id, name, currency, amount, close_date, ref)
except SourceRefTaken:
# Raced: another call claimed the ref between our lookup and our INSERT, and its transaction
# committed (that is what makes our INSERT collide). Ours rolled back with nothing written; the
# winner is read in a NEW transaction, because an errored one is aborted on Postgres and DataK3.
found = _opened_from(ref, bp_id)
if found is not None:
return found
raise Refused(409, f"a deal for {ref} is being opened by another request; try again") from None
def _opened_from(ref: str, bp_id: int) -> Optional[dict]:
"""The deal already opened for `ref` — or None if there is none yet. A deal for another partner is 409."""
with repository() as repo:
opportunity_id = repo.by_source(ref)
if opportunity_id is None:
return None
o = repo.get(opportunity_id)
if o is None:
return None
if o.bp_id != int(bp_id):
raise Refused(409, f"a deal was already opened for {ref}, for partner {o.bp_id} — not {bp_id}. "
f"Nothing was written")
return _view(o, *repo.pipeline())
@PIPE.use_case(kind="command", route="PATCH /opportunities/{opportunity_id}", permission=WRITE,
idempotent=True, if_match="version")
def edit_opportunity(caller: dict, opportunity_id: Ref, name: Optional[Name] = None,
amount: Optional[Amount] = None, close_date: Optional[IsoDate] = None,
probability: Optional[Percent] = None, use_stage_probability: bool = False,
version: Optional[int] = None) -> dict:
"""Change a deal's name, amount, close date or probability. Fields left out are unchanged. A closed deal's
amount and close date are frozen (a won one's are the contract's). `use_stage_probability` drops this
deal's own probability and goes back to its stage's. `version` (or If-Match): the one you read."""
def run():
with repository() as repo, _rules():
o = _at(_load(repo, caller, opportunity_id, write=True), version)
p, _ = repo.pipeline()
moved = []
if name is not None and name != o.name:
o.rename(name)
moved.append(f"name → {o.name}")
if amount is not None and (o.amount is None or o.amount.amount != Money.of(amount, o.currency).amount):
before = o.amount
o.set_amount(amount, p)
moved.append(f"amount {before} → {o.amount}")
if close_date is not None and close_date[:10] != (o.close_date or ""):
o.set_close_date(close_date, p)
moved.append(f"close date → {o.close_date}")
if use_stage_probability or probability is not None:
o.set_probability(None if use_stage_probability else probability, p)
moved.append(f"probability → {o.probability_for(p)}%")
if not moved:
return _view(o, p, repo.pipeline()[1])
return _changed(repo, caller, o, "edited", "; ".join(moved))
return _changing(version, run)
def _win_evidence(o: Opportunity, caller: dict) -> tuple:
"""(has_accepted, total, won_quote_ids, note) — what the quote brick says about THIS deal's win.
Only an acceptance by the deal's own customer, in its currency, and not voided by a reopen counts (C1): a
quote for another partner attached to this deal's id, before quote 0.2 checked, must not win it."""
if not QUOTES.installed:
return None, None, (), " (no quote module: won without an accepted-quote check)"
r = QUOTES.get(f"/opportunities/{o.opportunity_id}/quote-facts", actor=caller.get("email", ""))
if not r.ok:
raise Refused(503, f"cannot check this deal's quotes right now ({r.reason}); "
f"try again — a won deal is recorded only with its evidence")
accepted = r.data.get("accepted")
if accepted is None:
raise Refused(503, "the quote module cannot say which customer accepted (it is older than quote 0.2); "
"upgrade it — a won deal is recorded only with its evidence")
mine = [a for a in accepted if int(a["bp_id"]) == o.bp_id and a["currency"] == o.currency
and o.counts_as_evidence(str(a["quote_id"]))]
ignored = [str(a["quote_id"]) for a in accepted if a not in mine]
note = f" (not counted: {', '.join(ignored)} — another customer, currency or a voided win)" if ignored else ""
if not mine:
return False, None, (), note
total = Money.zero(o.currency)
for a in mine:
total = total + Money.of(a["total"], o.currency)
return True, total, tuple(str(a["quote_id"]) for a in mine), note
@PIPE.use_case(kind="command", route="POST /opportunities/{opportunity_id}/stage", permission=WRITE,
idempotent=True, if_match="version")
def move_opportunity(caller: dict, opportunity_id: Ref, stage: StageName,
loss_reason: Optional[Reason] = None, version: Optional[int] = None) -> dict:
"""CRM-04: move a deal to a stage; its probability and forecast category follow the stage. The won
stage needs an accepted quote from the deal's customer, in its currency, and the deal then takes that
quote's total; a lost stage needs a reason from the agreed list. A closed deal is final — a manager
reopens one (reopen_opportunity). Moving a deal to the stage it is in changes nothing."""
def run():
with repository() as repo, _rules():
o = _at(_load(repo, caller, opportunity_id, write=True), version)
p, _ = repo.pipeline()
target = p.stage(stage)
if o.stage == target.name:
return _view(o, p, repo.pipeline()[1])
has_accepted, total, won, note = None, None, (), ""
if target.is_won and o.is_open(p):
has_accepted, total, won, note = _win_evidence(o, caller)
amount_before = o.amount
before = o.move(target.name, p, has_accepted_quote=has_accepted, loss_reason=loss_reason,
accepted_total=total, won_quotes=won)
why = f" ({o.loss_reason})" if o.loss_reason else ""
took = (f"; amount {amount_before} → {o.amount} from {', '.join(won)}"
if won and amount_before != o.amount else (f"; won with {', '.join(won)}" if won else ""))
return _changed(repo, caller, o, "moved", f"{before} → {o.stage}{why}{took}{note}")
return _changing(version, run)
@PIPE.use_case(kind="command", route="POST /opportunities/{opportunity_id}/reopen", permission=REOPEN,
if_match="version")
def reopen_opportunity(caller: dict, opportunity_id: Ref, reason: Note, stage: Optional[StageName] = None,
version: Optional[int] = None) -> dict:
"""A manager opens a closed (won or lost) deal again, saying why — into `stage`, or the last open stage.
Reopening a WON deal voids the quotes that won it: winning it again needs a new acceptance. The forecast
it was counted in changes, which is why it is a manager's call and recorded with the reason."""
def run():
with repository() as repo, _rules():
o = _at(_load(repo, caller, opportunity_id, write=True), version)
p, _ = repo.pipeline()
to = stage or [s for s in p.stages if not s.is_closed][-1].name
before, voided = o.reopen(to, reason, p)
what = f"{before} → {o.stage}: {reason.strip()}" + (f" (voided the win by {', '.join(voided)})"
if voided else "")
return _changed(repo, caller, o, "reopened", what)
return _changing(version, run)
@PIPE.use_case(kind="command", route="PUT /opportunities/{opportunity_id}/category", permission=OVERRIDE,
idempotent=True, if_match="version")
def override_category(caller: dict, opportunity_id: Ref, category: Optional[Category] = None,
reason: Optional[Note] = None, version: Optional[int] = None) -> dict:
"""CRM-09: call an open deal commit, best case, pipeline or omitted against its stage, with a reason
(it is the number leadership is given). Leave `category` out to go back to the stage's."""
def run():
with repository() as repo, _rules():
o = _at(_load(repo, caller, opportunity_id), version)
p, _ = repo.pipeline()
o.override_category(category, reason, p)
return _changed(repo, caller, o, "category",
f"{category or 'stage default'}" + (f": {o.category_reason}" if category else ""))
return _changing(version, run)
@PIPE.use_case(kind="command", route="POST /opportunities/{opportunity_id}/touch", permission=TOUCH,
idempotent=True)
def touch_opportunity(caller: dict, opportunity_id: Ref, at: IsoTime, what: Name = "activity") -> dict:
"""Report activity on a deal — called by the activity brick when a call, email or meeting is logged —
so a deal that is being worked is not flagged as gone quiet (CRM-16). The latest time wins; reporting
an older activity changes nothing."""
def run():
with repository() as repo, _rules():
o = repo.get(opportunity_id)
if o is None:
raise Refused(404, f"no opportunity {opportunity_id}")
when = _aware(at)
if o.last_activity_at and _aware(o.last_activity_at) >= when:
return {"opportunity_id": opportunity_id, "last_activity_at": o.last_activity_at}
o.last_activity_at = when.isoformat()
repo.save(o)
repo.record(o.opportunity_id, "activity", caller["email"], what)
return {"opportunity_id": opportunity_id, "last_activity_at": o.last_activity_at}
return retry_on_conflict(run)
# ── the forecast snapshot (CRM-15) ─────────────────────────────────────────────────────────
@PIPE.use_case(kind="command", route="POST /forecast/snapshots", permission=SNAPSHOT, idempotent=True)
def take_forecast_snapshot(caller: dict, snapshot_date: Optional[IsoDate] = None) -> dict:
"""Record the forecast as it stands, for a date (default today): one row per rep, quarter and
currency. Taking it again for the same date REPLACES it rather than doubling it. There is no
scheduler on the platform: call this nightly from whatever runs your jobs."""
on = (snapshot_date or _today())[:10]
with repository() as repo, _rules():
p, _ = repo.pipeline()
rows = [r for r in forecast(repo.deals(), p) if r["period"] is not None]
repo.save_snapshot(on, rows, caller["email"])
return {"snapshot_date": on, "rows": len(rows)}
# ── configuration: the customer's own sales process ────────────────────────────────────────
@PIPE.use_case(kind="command", route="PUT /pipeline", permission=STAGES, idempotent=True)
def set_pipeline(caller: dict, stages: list[dict], loss_reasons: Optional[list[str]] = None,
risk_quiet_days: int = 21, risk_from_probability: Decimal = Decimal("50")) -> dict:
"""Replace the sales stages (each: name, position, probability, category, is_won, is_closed), the loss
reasons, and when a deal counts as at risk. A stage that still holds deals cannot be removed — move
them first, or the pipeline would hold deals in a stage that no longer exists."""
with repository() as repo, _rules():
try:
new = Pipeline(tuple(Stage(str(s["name"]).strip().lower(), int(s["position"]),
Decimal(str(s["probability"])), str(s["category"]),
bool(s.get("is_won", False)), bool(s.get("is_closed", False)))
for s in stages),
tuple(r.strip().lower() for r in (loss_reasons or DEFAULT_PIPELINE.loss_reasons)
if r.strip()))
except (KeyError, ValueError, TypeError) as e:
raise Refused(422, f"each stage needs name, position, probability and category ({e})") from None
if risk_quiet_days < 1:
raise Refused(422, "risk_quiet_days is at least 1")
old, _ = repo.pipeline()
kept = {s.name for s in new.stages}
for s in old.stages:
if s.name not in kept and repo.count_in_stage(s.name):
raise Refused(409, f"stage {s.name} still holds {repo.count_in_stage(s.name)} deal(s); "
f"move them before removing it")
repo.set_pipeline(new, RiskPolicy(risk_quiet_days, Decimal(str(risk_from_probability))), caller["email"])
return {"stages": [s.name for s in new.stages], "loss_reasons": list(new.loss_reasons),
"risk": {"quiet_days": risk_quiet_days, "from_probability": Decimal(str(risk_from_probability))}}