"""The tables, and the code that turns rows into deals, the pipeline and snapshots and back."""
import uuid
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from sqlalchemy import (BigInteger, Boolean, Column, Date, Integer, MetaData, Numeric, String, Table,
TIMESTAMP, func, insert, select)
from sqlalchemy.exc import IntegrityError
from erp_platform.money import Money
from erp_platform.repository import Aggregate, Part, upsert
from brick_pipeline.domain.pipeline import (DEFAULT_PIPELINE, Opportunity, Pipeline, RiskPolicy,
Stage)
md = MetaData()
M, P = Numeric(18, 2), Numeric(5, 2)
opportunities = Table(
"pipeline_opportunities", md,
Column("opportunity_id", String, primary_key=True), Column("bp_id", BigInteger),
Column("name", String), Column("owner", String), Column("stage", String), Column("currency", String),
Column("amount", M), Column("close_date", Date), Column("probability", P),
Column("category_override", String), Column("category_reason", String), Column("loss_reason", String),
Column("last_activity_at", TIMESTAMP), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
# 0.3.0, at the end as their ALTERs say: the quotes that won the deal, those a reopen voided, the version
Column("won_quote_ids", String), Column("voided_quote_ids", String), Column("version", Integer))
events = Table("pipeline_events", md,
Column("opportunity_id", String, primary_key=True), Column("event_id", String, primary_key=True),
Column("action", String), Column("detail", String), Column("actor", String),
Column("event_at", TIMESTAMP))
settings = Table("pipeline_settings", md,
Column("pipeline_id", String, primary_key=True), Column("loss_reasons", String),
Column("risk_quiet_days", Integer), Column("risk_from_pct", P),
Column("updated_by", String), Column("updated_at", TIMESTAMP))
stages = Table("pipeline_stages", md,
Column("pipeline_id", String, primary_key=True), Column("name", String, primary_key=True),
Column("position", Integer), Column("probability", P), Column("category", String),
Column("is_won", Boolean), Column("is_closed", Boolean))
runs = Table("pipeline_snapshot_runs", md,
Column("snapshot_date", Date, primary_key=True), Column("taken_by", String),
Column("taken_at", TIMESTAMP), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
snapshot_rows = Table(
"pipeline_snapshot_rows", md,
Column("snapshot_date", Date, primary_key=True), Column("owner", String, primary_key=True),
Column("period", String, primary_key=True), Column("currency", String, primary_key=True),
Column("open_amount", M), Column("weighted_amount", M), Column("commit_amount", M),
Column("won_amount", M), Column("deals", Integer), Column("unforecastable", Integer))
source_refs = Table("pipeline_source_refs", md,
Column("source_ref", String, primary_key=True), Column("opportunity_id", String),
Column("created_at", TIMESTAMP))
OPPORTUNITY = Aggregate(Part(opportunities, ("opportunity_id",)), version="version")
PIPELINE = Aggregate(Part(settings, ("pipeline_id",)), Part(stages, ("pipeline_id", "name")))
SNAPSHOT = Aggregate(Part(runs, ("snapshot_date",)),
Part(snapshot_rows, ("snapshot_date", "owner", "period", "currency")))
class SourceRefTaken(Exception):
"""Another call claimed this source_ref first. Its transaction is aborted on Postgres/DataK3, so the
winner is read in a NEW one (see open_opportunity)."""
def _is_duplicate(exc: Exception) -> bool:
"""A primary-key collision, however the driver says it: IntegrityError (psycopg, SQLite) or SQLSTATE
23505 in the message (the CLI the live check runs through). The same test brick-activity uses."""
text = str(exc).lower()
return isinstance(exc, IntegrityError) or "23505" in text or "duplicate key" in text
def _now() -> datetime:
return datetime.now(timezone.utc)
def _iso(v) -> Optional[str]:
return v.isoformat() if hasattr(v, "isoformat") else (v or None)
def _d(v) -> Optional[date]:
return None if not v else (v if isinstance(v, date) and not isinstance(v, datetime)
else date.fromisoformat(str(v)[:10]))
def _ts(v) -> Optional[datetime]:
return None if not v else (v if isinstance(v, datetime)
else datetime.fromisoformat(str(v).replace("Z", "+00:00")))
def _dec(v) -> Optional[Decimal]:
return None if v is None else Decimal(str(v))
def _ids(v) -> tuple:
return tuple(x for x in (v or "").split(",") if x)
class PipelineRepository:
def __init__(self, conn):
self.conn = conn
# ── deals ────────────────────────────────────────────────────────────────────────────
def _deal(self, r) -> Opportunity:
return Opportunity(
opportunity_id=r["opportunity_id"], bp_id=int(r["bp_id"]), name=r["name"], owner=r["owner"],
stage=r["stage"], currency=r["currency"],
amount=None if r["amount"] is None else Money.of(r["amount"], r["currency"]),
close_date=_iso(_d(r["close_date"])), probability=_dec(r["probability"]),
category_override=r["category_override"], category_reason=r["category_reason"],
loss_reason=r["loss_reason"], last_activity_at=_iso(_ts(r["last_activity_at"])),
won_quote_ids=_ids(r.get("won_quote_ids")), voided_quote_ids=_ids(r.get("voided_quote_ids")),
version=int(r.get("version") or 0))
def get(self, opportunity_id: str) -> Optional[Opportunity]:
rows = OPPORTUNITY.load(self.conn, {"opportunity_id": opportunity_id})
return None if rows is None else self._deal(rows["pipeline_opportunities"][0])
def save(self, o: Opportunity) -> None:
"""Claims version n+1 against the version the deal was loaded at (0 for a new one): a racer that saved
in between makes this a Conflict, and nothing of it is written (erp_platform.repository)."""
o.version = OPPORTUNITY.save(self.conn, {"pipeline_opportunities": [{
"opportunity_id": o.opportunity_id, "bp_id": o.bp_id, "name": o.name, "owner": o.owner,
"stage": o.stage, "currency": o.currency,
"amount": None if o.amount is None else o.amount.amount, "close_date": _d(o.close_date),
"probability": o.probability, "category_override": o.category_override,
"category_reason": o.category_reason, "loss_reason": o.loss_reason,
"last_activity_at": _ts(o.last_activity_at), "won_quote_ids": ",".join(o.won_quote_ids),
"voided_quote_ids": ",".join(o.voided_quote_ids)}]}, expected_version=o.version)
def deals(self, owner: Optional[str] = None, bp_ids: Optional[list] = None) -> list:
stmt = select(opportunities)
if owner is not None:
stmt = stmt.where(opportunities.c.owner == owner)
if bp_ids is not None:
stmt = stmt.where(opportunities.c.bp_id.in_([int(b) for b in bp_ids]))
return sorted((self._deal(dict(r._mapping)) for r in self.conn.execute(stmt)),
key=lambda o: (o.close_date or "9999", o.opportunity_id))
def count_in_stage(self, stage: str) -> int:
return sum(1 for _ in self.conn.execute(select(opportunities.c.opportunity_id)
.where(opportunities.c.stage == stage)))
# ── one source document, one deal (0.2.0) ────────────────────────────────────────────
def by_source(self, source_ref: str) -> Optional[str]:
"""The deal already opened for this source document, if any (committed rows only)."""
r = self.conn.execute(select(source_refs.c.opportunity_id)
.where(source_refs.c.source_ref == source_ref)).first()
return None if r is None else r[0]
def claim_source(self, source_ref: str, opportunity_id: str) -> None:
"""A plain INSERT on purpose: the one write here that must FAIL on a duplicate, not update it.
Written before the deal, in the same transaction, so a loser writes nothing at all."""
try:
self.conn.execute(insert(source_refs).values(source_ref=source_ref, opportunity_id=opportunity_id,
created_at=_now()))
except Exception as e: # noqa: BLE001 — re-raised below
if _is_duplicate(e):
raise SourceRefTaken(source_ref) from None
raise
# ── the pipeline: configuration ──────────────────────────────────────────────────────
def pipeline(self) -> tuple:
"""(Pipeline, RiskPolicy) — the stored rows, or the documented defaults."""
rows = PIPELINE.load(self.conn, {"pipeline_id": "default"})
if rows is None or not rows["pipeline_stages"]:
return DEFAULT_PIPELINE, RiskPolicy()
s = rows["pipeline_settings"][0]
p = Pipeline(tuple(Stage(r["name"], int(r["position"]), _dec(r["probability"]), r["category"],
bool(r["is_won"]), bool(r["is_closed"])) for r in rows["pipeline_stages"]),
tuple(x for x in s["loss_reasons"].split(",") if x))
return p, RiskPolicy(int(s["risk_quiet_days"]), _dec(s["risk_from_pct"]))
def set_pipeline(self, p: Pipeline, risk: RiskPolicy, actor: str) -> None:
PIPELINE.save(self.conn, {
"pipeline_settings": [{"pipeline_id": "default", "loss_reasons": ",".join(p.loss_reasons),
"risk_quiet_days": risk.quiet_days, "risk_from_pct": risk.from_probability,
"updated_by": actor}],
"pipeline_stages": [{"pipeline_id": "default", "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]})
# ── CRM-15 snapshots ─────────────────────────────────────────────────────────────────
def save_snapshot(self, on: str, rows: list, actor: str) -> None:
"""One snapshot per date. Saving the same date again REPLACES it — including deleting rows for a
rep or period that no longer has deals, which a plain upsert would leave behind."""
SNAPSHOT.save(self.conn, {
"pipeline_snapshot_runs": [{"snapshot_date": _d(on), "taken_by": actor, "taken_at": _now()}],
"pipeline_snapshot_rows": [{"snapshot_date": _d(on), "owner": r["owner"], "period": r["period"],
"currency": r["currency"], "open_amount": r["open"],
"weighted_amount": r["weighted"], "commit_amount": r["commit"],
"won_amount": r["won"], "deals": r["deals"],
"unforecastable": r["unforecastable"]} for r in rows]})
def snapshot(self, on: str) -> Optional[dict]:
rows = SNAPSHOT.load(self.conn, {"snapshot_date": _d(on)})
if rows is None:
return None
run = rows["pipeline_snapshot_runs"][0]
return {"snapshot_date": _iso(_d(run["snapshot_date"])), "taken_by": run["taken_by"],
"taken_at": _iso(_ts(run["taken_at"])),
"rows": sorted(({"owner": r["owner"], "period": r["period"], "currency": r["currency"],
"open": _dec(r["open_amount"]), "weighted": _dec(r["weighted_amount"]),
"commit": _dec(r["commit_amount"]), "won": _dec(r["won_amount"]),
"deals": int(r["deals"]), "unforecastable": int(r["unforecastable"])}
for r in rows["pipeline_snapshot_rows"]),
key=lambda r: (r["period"], r["owner"], r["currency"]))}
# ── history ──────────────────────────────────────────────────────────────────────────
def record(self, opportunity_id: str, action: str, actor: str, detail: str = "") -> None:
upsert(self.conn, events, [{"opportunity_id": opportunity_id, "event_id": uuid.uuid4().hex,
"action": action, "detail": detail, "actor": actor,
"event_at": _now()}], ("opportunity_id", "event_id"))
def history(self, opportunity_id: str) -> list:
rows = self.conn.execute(select(events.c.action, events.c.detail, events.c.actor, events.c.event_at)
.where(events.c.opportunity_id == opportunity_id))
return sorted(({"action": r[0], "detail": r[1], "actor": r[2], "at": _iso(_ts(r[3]))} for r in rows),
key=lambda e: e["at"] or "")