"""The tables, and the code that turns rows into a Quote and back.
The mapping is written by hand — the price of a domain with no SQLAlchemy in it. Every key value is
converted to the column's type HERE (a date as a `date`, not "2026-10"): the platform's repository
refuses a mistyped key, because a key that never compares equal to what is stored would make a save
delete the rows it just wrote.
"""
import uuid
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from sqlalchemy import (BigInteger, Column, Date, Integer, MetaData, Numeric, String, Table, TIMESTAMP, and_,
insert, select)
from sqlalchemy.exc import IntegrityError
from erp_platform.money import Money
from erp_platform.repository import Aggregate, Part, upsert
from brick_quote.domain.quote import (Approval, DiscountPolicy, Line, PricedAt, Quote, Version)
md = MetaData()
M, P = Numeric(18, 2), Numeric(5, 2)
quotes = Table("quote_quotes", md,
Column("quote_id", String, primary_key=True), Column("bp_id", BigInteger),
Column("opportunity_id", String), Column("currency", String), Column("owner", String),
Column("reference", String), Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
# 0.2.0: the optimistic version — "revision", because a quote's `version` is the offer the
# customer reads (v1, v2). At the end of the CREATE, as its ALTER says.
Column("revision", Integer))
versions = Table("quote_versions", md,
Column("quote_id", String, primary_key=True),
Column("version", Integer, primary_key=True), Column("status", String),
Column("valid_until", Date), Column("sent_at", TIMESTAMP),
Column("approved_by", String), Column("approved_pct", P),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
lines = Table("quote_lines", md,
Column("quote_id", String, primary_key=True), Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True), Column("product_id", String),
Column("description", String), Column("quantity", Integer), Column("list_price", M),
Column("unit_price", M), Column("currency", String), Column("price_book_id", String),
Column("priced_on", Date), Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
schedule = Table("quote_line_schedule", md,
Column("quote_id", String, primary_key=True),
Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("revenue_month", Date, primary_key=True), Column("amount", M))
events = Table("quote_events", md,
Column("quote_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))
policy = Table("quote_discount_policy", md,
Column("policy_id", String, primary_key=True), Column("auto_approve_pct", P),
Column("max_pct", P), Column("updated_by", String), Column("updated_at", TIMESTAMP))
prices = Table("quote_prices", md,
Column("price_book_id", String, primary_key=True),
Column("product_id", String, primary_key=True),
Column("currency", String, primary_key=True),
Column("valid_from", Date, primary_key=True), Column("list_price", M),
Column("description", String), Column("updated_by", String),
Column("updated_at", TIMESTAMP))
# 0.2.0 (H4): one quote in flight per deal, held under concurrency. The n-th quote opened on a deal claims
# (opportunity_id, n) with a plain INSERT: two opens that both saw n-1 quotes collide, and the loser answers the
# winner's quote instead of making a second one.
deal_quotes = Table("quote_deal_quotes", md,
Column("opportunity_id", String, primary_key=True),
Column("seq", Integer, primary_key=True, autoincrement=False),
Column("quote_id", String), Column("created_at", TIMESTAMP))
QUOTE = Aggregate(Part(quotes, ("quote_id",)),
Part(versions, ("quote_id", "version")),
Part(lines, ("quote_id", "version", "line_no")),
Part(schedule, ("quote_id", "version", "line_no", "revenue_month")),
version="revision")
class DealSlotTaken(Exception):
"""Another open claimed this deal's next quote first."""
def lost_deal_slot(exc: BaseException) -> bool:
"""This open lost the race for a deal's next quote — at the INSERT, or as a duplicate on quote_deal_quotes
when its transaction committed (DataK3 checks a staged key against another transaction's at COMMIT)."""
seen = set()
while exc is not None and id(exc) not in seen:
seen.add(id(exc))
if isinstance(exc, DealSlotTaken):
return True
text = str(exc).lower()
if (isinstance(exc, IntegrityError) or "23505" in text or "duplicate key" in text) and \
"quote_deal_quotes" in text:
return True
exc = getattr(exc, "orig", None) or exc.__cause__ or exc.__context__
return False
def _d(v) -> Optional[date]:
if v in (None, ""):
return None
return v if isinstance(v, date) and not isinstance(v, datetime) else date.fromisoformat(str(v)[:10])
def _iso(v) -> Optional[str]:
return v.isoformat() if hasattr(v, "isoformat") else (v or None)
def _ts(v) -> Optional[datetime]:
if v in (None, ""):
return None
return v if isinstance(v, datetime) else datetime.fromisoformat(str(v).replace("Z", "+00:00"))
class QuoteRepository:
"""Loads and saves a Quote whole, on a connection the use case owns (one transaction)."""
def __init__(self, conn):
self.conn = conn
# ── the aggregate ────────────────────────────────────────────────────────────────────
def get(self, quote_id: str) -> Optional[Quote]:
rows = QUOTE.load(self.conn, {"quote_id": quote_id})
if rows is None:
return None
head = rows["quote_quotes"][0]
months: dict = {}
for r in rows["quote_line_schedule"]:
months.setdefault((r["version"], r["line_no"]), []).append(
_d(r["revenue_month"]).isoformat()[:7])
by_version: dict = {}
for r in rows["quote_lines"]:
cur = r["currency"]
by_version.setdefault(r["version"], []).append(Line(
line_no=r["line_no"], product_id=r["product_id"], quantity=r["quantity"],
list_price=Money.of(r["list_price"], cur), unit_price=Money.of(r["unit_price"], cur),
priced_at=PricedAt(r["price_book_id"], _iso(_d(r["priced_on"]))),
description=r["description"] or "",
revenue_months=tuple(months.get((r["version"], r["line_no"]), ()))))
q = Quote(quote_id=head["quote_id"], bp_id=int(head["bp_id"]),
currency=head["currency"], owner=head["owner"],
opportunity_id=head["opportunity_id"], reference=head["reference"],
revision=int(head.get("revision") or 0))
for r in sorted(rows["quote_versions"], key=lambda r: r["version"]):
q.versions.append(Version(
version=r["version"], status=r["status"],
lines=sorted(by_version.get(r["version"], []), key=lambda ln: ln.line_no),
valid_until=_iso(_d(r["valid_until"])), sent_at=_iso(_ts(r["sent_at"])),
approval=(Approval(r["approved_by"], Decimal(str(r["approved_pct"])))
if r["approved_by"] else None)))
return q
def save(self, q: Quote) -> None:
"""The whole quote, claiming revision n+1 against the revision it was loaded at (0 for a new one): a
racer that saved in between makes this a Conflict, and nothing of it is written."""
q.revision = QUOTE.save(self.conn, {
"quote_quotes": [{"quote_id": q.quote_id, "bp_id": q.bp_id,
"opportunity_id": q.opportunity_id, "currency": q.currency,
"owner": q.owner, "reference": q.reference}],
"quote_versions": [{"quote_id": q.quote_id, "version": v.version, "status": v.status,
"valid_until": _d(v.valid_until), "sent_at": _ts(v.sent_at),
"approved_by": v.approval.approved_by if v.approval else None,
"approved_pct": v.approval.approved_pct if v.approval else None}
for v in q.versions],
"quote_lines": [{"quote_id": q.quote_id, "version": v.version, "line_no": ln.line_no,
"product_id": ln.product_id, "description": ln.description,
"quantity": ln.quantity, "list_price": ln.list_price.amount,
"unit_price": ln.unit_price.amount, "currency": ln.currency,
"price_book_id": ln.priced_at.price_book_id,
"priced_on": _d(ln.priced_at.priced_on)}
for v in q.versions for ln in v.lines],
"quote_line_schedule": [{"quote_id": q.quote_id, "version": v.version,
"line_no": ln.line_no, "revenue_month": _d(f"{m}-01"),
"amount": amt.amount}
for v in q.versions for ln in v.lines
for m, amt in ln.schedule().items()]}, expected_version=q.revision)
def claim_deal_slot(self, opportunity_id: str, seq: int, quote_id: str) -> None:
"""A plain INSERT on purpose: the write that must FAIL when another open took the deal's n-th quote."""
try:
self.conn.execute(insert(deal_quotes).values(opportunity_id=opportunity_id, seq=seq, quote_id=quote_id,
created_at=datetime.now(timezone.utc)))
except Exception as e: # noqa: BLE001 — re-raised below
if lost_deal_slot(e) or isinstance(e, IntegrityError):
raise DealSlotTaken(opportunity_id) from None
raise
# ── reads beside the aggregate ───────────────────────────────────────────────────────
def ids(self, owner: Optional[str] = None, opportunity_id: Optional[str] = None) -> list:
stmt = select(quotes.c.quote_id)
if owner is not None:
stmt = stmt.where(quotes.c.owner == owner)
if opportunity_id is not None:
stmt = stmt.where(quotes.c.opportunity_id == opportunity_id)
return sorted(r[0] for r in self.conn.execute(stmt))
def history(self, quote_id: str) -> list:
rows = self.conn.execute(select(events.c.action, events.c.detail, events.c.actor,
events.c.event_at).where(events.c.quote_id == quote_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 "")
def record(self, quote_id: str, action: str, actor: str, detail: str = "",
now: Optional[datetime] = None) -> None:
"""Append one history row. A fresh id each time: history is appended, never merged."""
upsert(self.conn, events, [{"quote_id": quote_id, "event_id": uuid.uuid4().hex,
"action": action, "detail": detail, "actor": actor,
"event_at": now or datetime.now(timezone.utc)}],
("quote_id", "event_id"))
# ── configuration rows ───────────────────────────────────────────────────────────────
def policy(self) -> DiscountPolicy:
"""The 'default' policy row, or the documented defaults when none has been set."""
r = self.conn.execute(select(policy.c.auto_approve_pct, policy.c.max_pct)
.where(policy.c.policy_id == "default")).first()
return DiscountPolicy(Decimal(str(r[0])), Decimal(str(r[1]))) if r else DiscountPolicy()
def set_policy(self, p: DiscountPolicy, actor: str) -> None:
upsert(self.conn, policy, [{"policy_id": "default", "auto_approve_pct": p.auto_approve_pct,
"max_pct": p.max_pct, "updated_by": actor,
"updated_at": datetime.now(timezone.utc)}], "policy_id")
def price(self, price_book_id: str, product_id: str, currency: str, on: str) -> Optional[dict]:
"""The list price in force on `on`: the latest row whose valid_from is on or before it."""
rows = self.conn.execute(select(prices.c.valid_from, prices.c.list_price, prices.c.description)
.where(and_(prices.c.price_book_id == price_book_id,
prices.c.product_id == product_id,
prices.c.currency == currency,
prices.c.valid_from <= _d(on))))
best = max(rows, key=lambda r: _d(r[0]), default=None)
if best is None:
return None
return {"valid_from": _iso(_d(best[0])), "list_price": Decimal(str(best[1])),
"description": best[2] or ""}
def price_at(self, price_book_id: str, product_id: str, currency: str,
valid_from: str) -> Optional[Decimal]:
r = self.conn.execute(select(prices.c.list_price).where(and_(
prices.c.price_book_id == price_book_id, prices.c.product_id == product_id,
prices.c.currency == currency, prices.c.valid_from == _d(valid_from)))).first()
return Decimal(str(r[0])) if r else None
def set_price(self, price_book_id: str, product_id: str, currency: str, valid_from: str,
list_price: Decimal, description: str, actor: str) -> None:
upsert(self.conn, prices, [{"price_book_id": price_book_id, "product_id": product_id,
"currency": currency, "valid_from": _d(valid_from),
"list_price": list_price, "description": description,
"updated_by": actor, "updated_at": datetime.now(timezone.utc)}],
("price_book_id", "product_id", "currency", "valid_from"))