whqreza
8adfd2267f
Materials are now global and category-free. The category lives only on the supplier (suppliers.category_id); every category-facing query goes through the supplier edge instead of the material row.
Schema (migration.sql + init_db DDL):
- drop the materials.category_id column, its FK, and the v2.5 first-link-wins automation (fn_derive_material_category + trg_supplier_materials_derive_category) — the DDL runs idempotently at API startup
Backend:
- GET /materials?category_id=… now filters via EXISTS(supplier_materials ⋈ suppliers.category_id)
- quote category filters (quotes list) use the quote's supplier category
- /pm/on-the-way derives the display category from the supplier of the quote
- supplier_materials links no longer derive/return a material category
- expand_service: material expand=category resolves to null (kept for contract compatibility)
- material create/update no longer validates or stores category_id; seed.py aligned
Frontend:
- material creation no longer sends category_id
- formatMaterialTitle reworked to a strict order with « . » separator: [Brand] . [Material] . [Spec 1] . [Spec 2], brand deduped when the stored name already starts with it; specs separator switched from « · » to « . »
Validated: python compile, api + frontend image builds (tsc + vite) clean; branch cut from the tip of prod_v4 (= origin/production, 5a9ef40 — the team's merge made them identical).
704 lines
28 KiB
Python
704 lines
28 KiB
Python
"""Expand-aware serialization.
|
|
|
|
Given a set of expand tokens, serializers resolve FK IDs into the nested
|
|
object shapes defined by the contracts (expert-opinion-p0/p1/p2).
|
|
|
|
All list serializers batch-load their dependencies (no N+1): one round of
|
|
queries per related table, then assembly from in-memory maps.
|
|
"""
|
|
import uuid
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.catalog import (
|
|
AttributeDefinition,
|
|
Brand,
|
|
BrandMaterial,
|
|
Material,
|
|
MaterialAttributeValue,
|
|
MaterialCategory,
|
|
Unit,
|
|
)
|
|
from app.models.procurement import (
|
|
ItemDecision,
|
|
ItemReceipt,
|
|
ItemReceiptAllocation,
|
|
ItemTermsheet,
|
|
LineItem,
|
|
LineItemProjects,
|
|
LineItemSpec,
|
|
Project,
|
|
Quote,
|
|
QuoteEvent,
|
|
QuoteProjects,
|
|
Supplier,
|
|
Termsheet,
|
|
TermsheetScheduleStep,
|
|
)
|
|
from app.models.inventory import Warehouse
|
|
from app.services.eav_service import flatten_eav_values
|
|
from app.utils.serialization import to_date, to_iso
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Small helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def parse_expand(expand: str | None) -> set[str]:
|
|
if not expand:
|
|
return set()
|
|
return {token.strip() for token in expand.split(",") if token.strip()}
|
|
|
|
|
|
def unit_to_dict(unit: Unit | None) -> dict | None:
|
|
if unit is None:
|
|
return None
|
|
return {"id": unit.id, "code": unit.code, "name_fa": unit.name_fa}
|
|
|
|
|
|
def category_brief(category: MaterialCategory | None) -> dict | None:
|
|
if category is None:
|
|
return None
|
|
return {"id": str(category.id), "slug": category.slug, "name": category.name}
|
|
|
|
|
|
def supplier_brief(supplier: Supplier | None) -> dict | None:
|
|
if supplier is None:
|
|
return None
|
|
return {"id": str(supplier.id), "name": supplier.name, "location": supplier.location}
|
|
|
|
|
|
def project_brief(project: Project | None) -> dict | None:
|
|
if project is None:
|
|
return None
|
|
return {"id": str(project.id), "name": project.name}
|
|
|
|
|
|
def decision_brief(d: ItemDecision, project: Project | None = None) -> dict:
|
|
"""Shape used inside expanded line items (P0-14)."""
|
|
return {
|
|
"id": str(d.id),
|
|
"status": d.status,
|
|
"chosen_item_termsheet_id": str(d.chosen_item_termsheet_id) if d.chosen_item_termsheet_id else None,
|
|
"project_id": str(d.project_id) if d.project_id else None, # v2.6: the single purchased project
|
|
"project": project_brief(project) if project is not None else None,
|
|
"adjusted_unit_price": float(d.adjusted_unit_price) if d.adjusted_unit_price is not None else None,
|
|
"decided_at": to_iso(d.decided_at),
|
|
}
|
|
|
|
|
|
async def _attach_quote_projects(db: AsyncSession, quotes: list[Quote], out: dict) -> None:
|
|
"""Batch-load quote_projects junction rows and attach a `projects` array."""
|
|
quote_ids = [q.id for q in quotes]
|
|
qp_rows = (
|
|
(
|
|
await db.execute(
|
|
select(QuoteProjects).where(QuoteProjects.quote_id.in_(quote_ids))
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
) if quote_ids else []
|
|
proj_ids = {r.project_id for r in qp_rows}
|
|
proj_map: dict = {}
|
|
if proj_ids:
|
|
projs = (await db.execute(select(Project).where(Project.id.in_(proj_ids)))).scalars().all()
|
|
proj_map = {p.id: p for p in projs}
|
|
projects_by_quote: dict = {}
|
|
for r in qp_rows:
|
|
projects_by_quote.setdefault(r.quote_id, []).append(proj_map[r.project_id])
|
|
for q in quotes:
|
|
out[q.id]["projects"] = [
|
|
project_brief(p) for p in projects_by_quote.get(q.id, []) if p is not None
|
|
]
|
|
|
|
|
|
def decision_full(d: ItemDecision) -> dict:
|
|
"""Shape used by the decision endpoints (P1-21)."""
|
|
return {
|
|
"id": str(d.id),
|
|
"line_item_id": str(d.line_item_id),
|
|
"chosen_item_termsheet_id": str(d.chosen_item_termsheet_id) if d.chosen_item_termsheet_id else None,
|
|
"status": d.status,
|
|
"project_id": str(d.project_id) if d.project_id else None, # v2.6: the single purchased project
|
|
"adjusted_unit_price": float(d.adjusted_unit_price) if d.adjusted_unit_price is not None else None,
|
|
"adjusted_notes": d.adjusted_notes,
|
|
"decided_at": to_iso(d.decided_at),
|
|
}
|
|
|
|
|
|
def receipt_brief(r: ItemReceipt) -> dict:
|
|
"""Shape used inside expanded line items (P0-14)."""
|
|
return {
|
|
"id": str(r.id),
|
|
"received_qty": float(r.received_qty),
|
|
"received_at": to_iso(r.received_at),
|
|
"on_time": r.on_time,
|
|
}
|
|
|
|
|
|
def receipt_full(r: ItemReceipt) -> dict:
|
|
"""Shape used by the receipt endpoints (P1-25)."""
|
|
return {
|
|
"id": str(r.id),
|
|
"item_decision_id": str(r.item_decision_id),
|
|
"received_at": to_iso(r.received_at),
|
|
"received_qty": float(r.received_qty),
|
|
"on_time": r.on_time,
|
|
"quality_note": r.quality_note,
|
|
}
|
|
|
|
|
|
def event_to_dict(e: QuoteEvent) -> dict:
|
|
return {
|
|
"id": str(e.id),
|
|
"quote_id": str(e.quote_id),
|
|
"event_type": e.event_type,
|
|
"payload": e.payload,
|
|
"occurred_at": to_iso(e.occurred_at),
|
|
"actor": e.actor,
|
|
}
|
|
|
|
|
|
def schedule_step_to_dict(s: TermsheetScheduleStep) -> dict:
|
|
return {
|
|
"id": str(s.id),
|
|
"step_no": s.step_no,
|
|
"due_date": to_date(s.due_date),
|
|
"amount": float(s.amount),
|
|
"note": s.note,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Quote (flat)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def quote_flat(q: Quote) -> dict:
|
|
d = {
|
|
"id": str(q.id),
|
|
"supplier_id": str(q.supplier_id),
|
|
"project_id": str(q.project_id) if q.project_id else None,
|
|
"quote_date": to_iso(q.quote_date),
|
|
"expiry_date": to_date(q.expiry_date),
|
|
"accepted_date": to_iso(q.accepted_date),
|
|
"arrived_date": to_iso(q.arrived_date),
|
|
"status": q.status,
|
|
"is_official": q.is_official,
|
|
"notes": q.notes,
|
|
"created_at": to_iso(q.created_at),
|
|
"updated_at": to_iso(q.updated_at),
|
|
}
|
|
# Lazy auto-expiry flag (contract P0-14): never changes the stored status.
|
|
if q.status == "offered" and q.expiry_date is not None and q.expiry_date < date.today():
|
|
d["is_expired"] = True
|
|
return d
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Materials
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _load_material_maps(db: AsyncSession, materials: list[Material], expand: set[str]) -> dict:
|
|
material_ids = [m.id for m in materials]
|
|
maps: dict = {"categories": {}, "units": {}, "brands": {}, "material_brands": {}, "mavs": {}, "attr_defs": {}}
|
|
|
|
unit_ids: set[int] = set()
|
|
if "canonical_unit" in expand:
|
|
unit_ids |= {m.canonical_unit_id for m in materials}
|
|
|
|
if "brands" in expand:
|
|
bms = (await db.execute(select(BrandMaterial).where(BrandMaterial.material_id.in_(material_ids)))).scalars().all()
|
|
by_material: dict[uuid.UUID, list[BrandMaterial]] = {}
|
|
brand_ids: set[uuid.UUID] = set()
|
|
for bm in bms:
|
|
by_material.setdefault(bm.material_id, []).append(bm)
|
|
brand_ids.add(bm.brand_id)
|
|
maps["material_brands"] = by_material
|
|
if brand_ids:
|
|
brands = (await db.execute(select(Brand).where(Brand.id.in_(brand_ids)))).scalars().all()
|
|
maps["brands"] = {b.id: b for b in brands}
|
|
|
|
if "attributes" in expand:
|
|
mavs = (
|
|
await db.execute(
|
|
select(MaterialAttributeValue)
|
|
.where(MaterialAttributeValue.material_id.in_(material_ids))
|
|
.order_by(MaterialAttributeValue.attribute_definition_id)
|
|
)
|
|
).scalars().all()
|
|
maps["mavs"] = mavs
|
|
def_ids = {row.attribute_definition_id for row in mavs}
|
|
if def_ids:
|
|
defs = (await db.execute(select(AttributeDefinition).where(AttributeDefinition.id.in_(def_ids)))).scalars().all()
|
|
maps["attr_defs"] = {d.id: d for d in defs}
|
|
unit_ids |= {d.unit_id for d in defs if d.unit_id is not None}
|
|
|
|
if unit_ids:
|
|
units = (await db.execute(select(Unit).where(Unit.id.in_(unit_ids)))).scalars().all()
|
|
maps["units"] = {u.id: u for u in units}
|
|
return maps
|
|
|
|
|
|
def _material_to_dict(m: Material, expand: set[str], maps: dict) -> dict:
|
|
d = {
|
|
"id": str(m.id),
|
|
"name": m.name,
|
|
"topic": m.topic,
|
|
"status": m.status,
|
|
"created_at": to_iso(m.created_at),
|
|
"updated_at": to_iso(m.updated_at),
|
|
}
|
|
if "category" in expand:
|
|
d["category"] = None
|
|
else:
|
|
d["category_id"] = None
|
|
if "canonical_unit" in expand:
|
|
d["canonical_unit"] = unit_to_dict(maps["units"].get(m.canonical_unit_id))
|
|
else:
|
|
d["canonical_unit_id"] = m.canonical_unit_id
|
|
if "brands" in expand:
|
|
d["brands"] = [
|
|
{
|
|
"id": str(bm.brand_id),
|
|
"name": maps["brands"][bm.brand_id].name,
|
|
"catalog_code": bm.catalog_code,
|
|
"is_active": bm.is_active,
|
|
}
|
|
for bm in maps["material_brands"].get(m.id, [])
|
|
if bm.brand_id in maps["brands"]
|
|
]
|
|
if "attributes" in expand:
|
|
rows = [row for row in maps["mavs"] if row.material_id == m.id]
|
|
d["attributes"] = flatten_eav_values(rows, maps["attr_defs"], maps["units"], prefer_unit_id_at_time=False)
|
|
return d
|
|
|
|
|
|
async def serialize_materials(db: AsyncSession, materials: list[Material], expand: set[str]) -> list[dict]:
|
|
if not materials:
|
|
return []
|
|
maps = await _load_material_maps(db, materials, expand)
|
|
return [_material_to_dict(m, expand, maps) for m in materials]
|
|
|
|
|
|
async def serialize_material(db: AsyncSession, material: Material, expand: set[str]) -> dict:
|
|
return (await serialize_materials(db, [material], expand))[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Line items
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _LineItemData:
|
|
"""Batch-loaded dependency maps for a set of line items."""
|
|
|
|
def __init__(self):
|
|
self.materials: dict[uuid.UUID, Material] = {}
|
|
self.categories: dict[uuid.UUID, MaterialCategory] = {}
|
|
self.units: dict[int, Unit] = {}
|
|
self.brands: dict[uuid.UUID, Brand] = {}
|
|
self.material_brands: dict[uuid.UUID, list[BrandMaterial]] = {}
|
|
self.specs_by_line: dict[uuid.UUID, list[LineItemSpec]] = {}
|
|
self.attr_defs: dict[uuid.UUID, AttributeDefinition] = {}
|
|
self.decisions: dict[uuid.UUID, ItemDecision] = {}
|
|
self.receipts_by_decision: dict[uuid.UUID, list[ItemReceipt]] = {}
|
|
self.allocations_by_receipt: dict[uuid.UUID, list[dict]] = {}
|
|
self.item_termsheets_by_line: dict[uuid.UUID, list[ItemTermsheet]] = {}
|
|
self.item_termsheets_by_id: dict[uuid.UUID, ItemTermsheet] = {}
|
|
self.projects_by_line: dict[uuid.UUID, list[Project]] = {}
|
|
self.projects_by_id: dict[uuid.UUID, Project] = {}
|
|
|
|
|
|
async def _load_line_item_data(db: AsyncSession, line_items: list[LineItem], expand: set[str]) -> _LineItemData:
|
|
data = _LineItemData()
|
|
li_ids = [li.id for li in line_items]
|
|
unit_ids: set[int] = {li.unit_id for li in line_items}
|
|
|
|
if "material" in expand:
|
|
material_ids = {li.material_id for li in line_items}
|
|
materials = (await db.execute(select(Material).where(Material.id.in_(material_ids)))).scalars().all()
|
|
data.materials = {m.id: m for m in materials}
|
|
bms = (await db.execute(select(BrandMaterial).where(BrandMaterial.material_id.in_(material_ids)))).scalars().all()
|
|
brand_ids: set[uuid.UUID] = set()
|
|
for bm in bms:
|
|
data.material_brands.setdefault(bm.material_id, []).append(bm)
|
|
brand_ids.add(bm.brand_id)
|
|
if brand_ids:
|
|
brands = (await db.execute(select(Brand).where(Brand.id.in_(brand_ids)))).scalars().all()
|
|
data.brands = {b.id: b for b in brands}
|
|
|
|
# Pricing always needs item_termsheets in expanded mode.
|
|
its = (await db.execute(select(ItemTermsheet).where(ItemTermsheet.line_item_id.in_(li_ids)))).scalars().all()
|
|
for it in its:
|
|
data.item_termsheets_by_line.setdefault(it.line_item_id, []).append(it)
|
|
data.item_termsheets_by_id[it.id] = it
|
|
|
|
# Per-material project allocation — always attached in expanded mode.
|
|
lip_rows = (
|
|
(await db.execute(select(LineItemProjects).where(LineItemProjects.line_item_id.in_(li_ids)))).scalars().all()
|
|
) if li_ids else []
|
|
proj_ids = {r.project_id for r in lip_rows}
|
|
proj_map: dict[uuid.UUID, Project] = {}
|
|
if proj_ids:
|
|
projs = (await db.execute(select(Project).where(Project.id.in_(proj_ids)))).scalars().all()
|
|
proj_map = {p.id: p for p in projs}
|
|
for r in lip_rows:
|
|
if r.project_id in proj_map:
|
|
data.projects_by_line.setdefault(r.line_item_id, []).append(proj_map[r.project_id])
|
|
|
|
if "specs" in expand:
|
|
specs = (
|
|
await db.execute(
|
|
select(LineItemSpec)
|
|
.where(LineItemSpec.line_item_id.in_(li_ids))
|
|
.order_by(LineItemSpec.attribute_definition_id)
|
|
)
|
|
).scalars().all()
|
|
for spec in specs:
|
|
data.specs_by_line.setdefault(spec.line_item_id, []).append(spec)
|
|
def_ids = {spec.attribute_definition_id for spec in specs}
|
|
if def_ids:
|
|
defs = (await db.execute(select(AttributeDefinition).where(AttributeDefinition.id.in_(def_ids)))).scalars().all()
|
|
data.attr_defs = {d.id: d for d in defs}
|
|
unit_ids |= {d.unit_id for d in defs if d.unit_id is not None}
|
|
unit_ids |= {spec.unit_id_at_time for spec in specs if spec.unit_id_at_time is not None}
|
|
|
|
if "decision" in expand or "receipt" in expand:
|
|
decisions = (await db.execute(select(ItemDecision).where(ItemDecision.line_item_id.in_(li_ids)))).scalars().all()
|
|
data.decisions = {d.line_item_id: d for d in decisions}
|
|
dec_proj_ids = {d.project_id for d in decisions if d.project_id is not None}
|
|
missing_pids = dec_proj_ids - set(proj_map.keys())
|
|
if missing_pids:
|
|
extra_projs = (await db.execute(select(Project).where(Project.id.in_(missing_pids)))).scalars().all()
|
|
for p in extra_projs:
|
|
proj_map[p.id] = p
|
|
data.projects_by_id = proj_map
|
|
dec_ids = [d.id for d in decisions]
|
|
if dec_ids:
|
|
receipts = (await db.execute(select(ItemReceipt).where(ItemReceipt.item_decision_id.in_(dec_ids)))).scalars().all()
|
|
by_dec: dict[uuid.UUID, list[ItemReceipt]] = {}
|
|
for r in receipts:
|
|
by_dec.setdefault(r.item_decision_id, []).append(r)
|
|
data.receipts_by_decision = by_dec
|
|
|
|
# Warehouse allocations per receipt — needed by the ledger's
|
|
# «رسید شده» split (target warehouse per received part).
|
|
alloc_rows = (
|
|
await db.execute(
|
|
select(ItemReceiptAllocation, Warehouse)
|
|
.join(Warehouse, Warehouse.id == ItemReceiptAllocation.warehouse_id)
|
|
.where(
|
|
ItemReceiptAllocation.receipt_id.in_([r.id for r in receipts])
|
|
if receipts
|
|
else False
|
|
)
|
|
)
|
|
).all()
|
|
allocs_by_receipt: dict[uuid.UUID, list[dict]] = {}
|
|
for alloc, wh in alloc_rows:
|
|
allocs_by_receipt.setdefault(alloc.receipt_id, []).append(
|
|
{"warehouse_id": str(wh.id), "warehouse_name": wh.name, "quantity": float(alloc.quantity)}
|
|
)
|
|
data.allocations_by_receipt = allocs_by_receipt
|
|
|
|
if unit_ids:
|
|
units = (await db.execute(select(Unit).where(Unit.id.in_(unit_ids)))).scalars().all()
|
|
data.units = {u.id: u for u in units}
|
|
return data
|
|
|
|
|
|
def _line_item_flat(li: LineItem) -> dict:
|
|
return {
|
|
"id": str(li.id),
|
|
"quote_id": str(li.quote_id),
|
|
"material_id": str(li.material_id),
|
|
"brand_material_id": str(li.brand_material_id) if li.brand_material_id else None,
|
|
"unit_id": li.unit_id,
|
|
"quantity": float(li.quantity),
|
|
"notes": li.notes,
|
|
"created_at": to_iso(li.created_at),
|
|
}
|
|
|
|
|
|
def _price_source(li: LineItem, data: _LineItemData) -> ItemTermsheet | None:
|
|
"""unit_price source: the item's chosen termsheet link, else first link (P0-14)."""
|
|
decision = data.decisions.get(li.id)
|
|
if decision is not None and decision.chosen_item_termsheet_id is not None:
|
|
chosen = data.item_termsheets_by_id.get(decision.chosen_item_termsheet_id)
|
|
if chosen is not None:
|
|
return chosen
|
|
links = data.item_termsheets_by_line.get(li.id, [])
|
|
return links[0] if links else None
|
|
|
|
|
|
def _line_item_expanded(li: LineItem, expand: set[str], data: _LineItemData, include_termsheet_ids: bool) -> dict:
|
|
d: dict = {"id": str(li.id)}
|
|
if "material" in expand:
|
|
m = data.materials.get(li.material_id)
|
|
if m is not None:
|
|
d["material"] = {
|
|
"id": str(m.id),
|
|
"name": m.name,
|
|
"category": None,
|
|
"brands": [
|
|
{"id": str(bm.brand_id), "name": data.brands[bm.brand_id].name, "catalog_code": bm.catalog_code}
|
|
for bm in data.material_brands.get(m.id, [])
|
|
if bm.brand_id in data.brands
|
|
],
|
|
}
|
|
else:
|
|
d["material_id"] = str(li.material_id)
|
|
else:
|
|
d["material_id"] = str(li.material_id)
|
|
d["brand_material_id"] = str(li.brand_material_id) if li.brand_material_id else None
|
|
if "unit" in expand:
|
|
d["unit"] = unit_to_dict(data.units.get(li.unit_id))
|
|
else:
|
|
d["unit_id"] = li.unit_id
|
|
d["quantity"] = float(li.quantity)
|
|
|
|
price_it = _price_source(li, data)
|
|
d["unit_price"] = float(price_it.unit_price) if price_it is not None else None
|
|
d["total_price"] = float(price_it.unit_price * li.quantity) if price_it is not None else None
|
|
d["projects"] = [project_brief(p) for p in data.projects_by_line.get(li.id, [])]
|
|
|
|
if "specs" in expand:
|
|
d["specs"] = flatten_eav_values(
|
|
data.specs_by_line.get(li.id, []), data.attr_defs, data.units, include_data_type=False
|
|
)
|
|
decision = data.decisions.get(li.id)
|
|
dec_proj = data.projects_by_id.get(decision.project_id) if decision and decision.project_id else None
|
|
d["target_project"] = project_brief(dec_proj) if dec_proj is not None else None
|
|
if "decision" in expand:
|
|
d["decision"] = decision_brief(decision, dec_proj) if decision is not None else None
|
|
# chosen_item_termsheet_id یک id از جدول item_termsheets است؛ برای مقایسه با
|
|
# termsheet_ids در جلوی برنامه، id ترمشیت واقعی هم ارسال میشود.
|
|
if decision is not None and decision.chosen_item_termsheet_id is not None:
|
|
chosen_it = data.item_termsheets_by_id.get(decision.chosen_item_termsheet_id)
|
|
if chosen_it is not None:
|
|
d["accepted_termsheet_id"] = str(chosen_it.termsheet_id)
|
|
if "receipt" in expand:
|
|
# 1:N partial deliveries — emit the aggregate (sum of received qty,
|
|
# latest receipt kept for the received_at contract) + the warehouse
|
|
# allocations so the ledger can render the «رسید شده» split.
|
|
rs = data.receipts_by_decision.get(decision.id) if decision is not None else None
|
|
if rs:
|
|
latest = rs[-1]
|
|
total = sum((r.received_qty for r in rs), Decimal("0"))
|
|
allocs: list[dict] = []
|
|
for r in rs:
|
|
allocs.extend(data.allocations_by_receipt.get(r.id, []))
|
|
d["receipt"] = {
|
|
**receipt_brief(latest),
|
|
"received_qty": float(total),
|
|
"receipt_count": len(rs),
|
|
"allocations": allocs,
|
|
}
|
|
else:
|
|
d["receipt"] = None
|
|
if include_termsheet_ids:
|
|
d["termsheet_ids"] = [str(it.termsheet_id) for it in data.item_termsheets_by_line.get(li.id, [])]
|
|
return d
|
|
|
|
|
|
async def serialize_line_items(
|
|
db: AsyncSession, line_items: list[LineItem], expand: set[str], include_termsheet_ids: bool = False
|
|
) -> list[dict]:
|
|
if not line_items:
|
|
return []
|
|
if not expand:
|
|
return [_line_item_flat(li) for li in line_items]
|
|
data = await _load_line_item_data(db, line_items, expand)
|
|
return [_line_item_expanded(li, expand, data, include_termsheet_ids) for li in line_items]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Termsheets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _termsheet_flat(ts: Termsheet) -> dict:
|
|
return {
|
|
"id": str(ts.id),
|
|
"quote_id": str(ts.quote_id),
|
|
"label": ts.label,
|
|
"status": ts.status,
|
|
"discount_type": ts.discount_type,
|
|
"discount_value": float(ts.discount_value),
|
|
"created_at": to_iso(ts.created_at),
|
|
}
|
|
|
|
|
|
def _termsheet_quote_shape(ts: Termsheet, steps: list[TermsheetScheduleStep]) -> dict:
|
|
"""Shape inside expanded quotes (P0-14): no quote_id/created_at, steps nested."""
|
|
return {
|
|
"id": str(ts.id),
|
|
"label": ts.label,
|
|
"status": ts.status,
|
|
"discount_type": ts.discount_type,
|
|
"discount_value": float(ts.discount_value),
|
|
"schedule_steps": [schedule_step_to_dict(s) for s in steps],
|
|
}
|
|
|
|
|
|
async def serialize_termsheets(db: AsyncSession, termsheets: list[Termsheet], expand: set[str]) -> list[dict]:
|
|
if not termsheets:
|
|
return []
|
|
result = [_termsheet_flat(ts) for ts in termsheets]
|
|
ts_ids = [ts.id for ts in termsheets]
|
|
|
|
if "schedule_steps" in expand:
|
|
steps = (
|
|
await db.execute(
|
|
select(TermsheetScheduleStep)
|
|
.where(TermsheetScheduleStep.termsheet_id.in_(ts_ids))
|
|
.order_by(TermsheetScheduleStep.step_no)
|
|
)
|
|
).scalars().all()
|
|
grouped: dict[uuid.UUID, list[TermsheetScheduleStep]] = {}
|
|
for s in steps:
|
|
grouped.setdefault(s.termsheet_id, []).append(s)
|
|
for ts in termsheets:
|
|
for item in result:
|
|
if item["id"] == str(ts.id):
|
|
item["schedule_steps"] = [schedule_step_to_dict(s) for s in grouped.get(ts.id, [])]
|
|
break
|
|
|
|
if "line_items" in expand:
|
|
its = (await db.execute(select(ItemTermsheet).where(ItemTermsheet.termsheet_id.in_(ts_ids)))).scalars().all()
|
|
li_ids = [it.line_item_id for it in its]
|
|
quantities: dict[uuid.UUID, float] = {}
|
|
if li_ids:
|
|
rows = (await db.execute(select(LineItem).where(LineItem.id.in_(li_ids)))).scalars().all()
|
|
quantities = {li.id: float(li.quantity) for li in rows}
|
|
grouped_its: dict[uuid.UUID, list[ItemTermsheet]] = {}
|
|
for it in its:
|
|
grouped_its.setdefault(it.termsheet_id, []).append(it)
|
|
for ts in termsheets:
|
|
links = grouped_its.get(ts.id, [])
|
|
items = []
|
|
for it in links:
|
|
qty = quantities.get(it.line_item_id, 0)
|
|
items.append(
|
|
{
|
|
"id": str(it.id),
|
|
"line_item_id": str(it.line_item_id),
|
|
"unit_price": float(it.unit_price),
|
|
"min_order_qty": float(it.min_order_qty) if it.min_order_qty is not None else None,
|
|
"total_price": float(it.unit_price * qty),
|
|
}
|
|
)
|
|
for item in result:
|
|
if item["id"] == str(ts.id):
|
|
item["line_items"] = items
|
|
break
|
|
return result
|
|
|
|
|
|
async def serialize_termsheet(db: AsyncSession, termsheet: Termsheet, expand: set[str]) -> dict:
|
|
return (await serialize_termsheets(db, [termsheet], expand))[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Quotes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def serialize_quotes(db: AsyncSession, quotes: list[Quote], expand: set[str]) -> list[dict]:
|
|
if not quotes:
|
|
return []
|
|
out: dict[uuid.UUID, dict] = {q.id: quote_flat(q) for q in quotes}
|
|
quote_ids = [q.id for q in quotes]
|
|
await _attach_quote_projects(db, quotes, out)
|
|
|
|
if "supplier" in expand:
|
|
sup_ids = {q.supplier_id for q in quotes}
|
|
sups = (await db.execute(select(Supplier).where(Supplier.id.in_(sup_ids)))).scalars().all()
|
|
sup_map = {s.id: s for s in sups}
|
|
for q in quotes:
|
|
out[q.id]["supplier"] = supplier_brief(sup_map.get(q.supplier_id))
|
|
|
|
if "project" in expand:
|
|
proj_ids = {q.project_id for q in quotes if q.project_id is not None}
|
|
proj_map: dict = {}
|
|
if proj_ids:
|
|
projs = (await db.execute(select(Project).where(Project.id.in_(proj_ids)))).scalars().all()
|
|
proj_map = {p.id: p for p in projs}
|
|
for q in quotes:
|
|
out[q.id]["project"] = project_brief(proj_map.get(q.project_id)) if q.project_id else None
|
|
|
|
if "events" in expand:
|
|
events = (
|
|
await db.execute(
|
|
select(QuoteEvent)
|
|
.where(QuoteEvent.quote_id.in_(quote_ids))
|
|
.order_by(QuoteEvent.occurred_at.asc(), QuoteEvent.id)
|
|
)
|
|
).scalars().all()
|
|
grouped: dict[uuid.UUID, list[QuoteEvent]] = {}
|
|
for e in events:
|
|
grouped.setdefault(e.quote_id, []).append(e)
|
|
for q in quotes:
|
|
out[q.id]["events"] = [event_to_dict(e) for e in grouped.get(q.id, [])]
|
|
|
|
if "line_items" in expand:
|
|
items = (await db.execute(select(LineItem).where(LineItem.quote_id.in_(quote_ids)))).scalars().all()
|
|
grouped_li: dict[uuid.UUID, list[LineItem]] = {}
|
|
for li in items:
|
|
grouped_li.setdefault(li.quote_id, []).append(li)
|
|
for q in quotes:
|
|
lis = grouped_li.get(q.id, [])
|
|
if lis:
|
|
# Contract P0-14: expand=line_items implies full nested resolution —
|
|
# material (category + brands), unit, specs, decision, receipt,
|
|
# termsheet_ids — regardless of other tokens.
|
|
out[q.id]["line_items"] = await serialize_line_items(
|
|
db, lis, {"material", "unit", "specs", "decision", "receipt"}, include_termsheet_ids=True
|
|
)
|
|
else:
|
|
out[q.id]["line_items"] = []
|
|
|
|
if "termsheets" in expand:
|
|
termsheets = (
|
|
(await db.execute(select(Termsheet).where(Termsheet.quote_id.in_(quote_ids)))).scalars().all()
|
|
)
|
|
grouped_ts: dict[uuid.UUID, list[Termsheet]] = {}
|
|
for ts in termsheets:
|
|
grouped_ts.setdefault(ts.quote_id, []).append(ts)
|
|
all_ts_ids = [ts.id for ts in termsheets]
|
|
steps_by_ts: dict[uuid.UUID, list[TermsheetScheduleStep]] = {}
|
|
if all_ts_ids:
|
|
steps = (
|
|
await db.execute(
|
|
select(TermsheetScheduleStep)
|
|
.where(TermsheetScheduleStep.termsheet_id.in_(all_ts_ids))
|
|
.order_by(TermsheetScheduleStep.step_no)
|
|
)
|
|
).scalars().all()
|
|
for s in steps:
|
|
steps_by_ts.setdefault(s.termsheet_id, []).append(s)
|
|
for q in quotes:
|
|
out[q.id]["termsheets"] = [
|
|
_termsheet_quote_shape(ts, steps_by_ts.get(ts.id, [])) for ts in grouped_ts.get(q.id, [])
|
|
]
|
|
|
|
return [out[q.id] for q in quotes]
|
|
|
|
|
|
async def serialize_quote(db: AsyncSession, quote: Quote, expand: set[str]) -> dict:
|
|
return (await serialize_quotes(db, [quote], expand))[0]
|
|
|
|
|
|
async def serialize_quote_with_children(db: AsyncSession, quote: Quote) -> dict:
|
|
"""POST /quotes response shape: flat quote + flat line_items + flat termsheets."""
|
|
d = quote_flat(quote)
|
|
out = {quote.id: d}
|
|
await _attach_quote_projects(db, [quote], out)
|
|
d = out[quote.id]
|
|
lis = (await db.execute(select(LineItem).where(LineItem.quote_id == quote.id))).scalars().all()
|
|
d["line_items"] = [_line_item_flat(li) for li in lis]
|
|
tss = (await db.execute(select(Termsheet).where(Termsheet.quote_id == quote.id))).scalars().all()
|
|
d["termsheets"] = [_termsheet_flat(ts) for ts in tss]
|
|
return d
|