"""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 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, ItemTermsheet, LineItem, LineItemProjects, LineItemSpec, Project, Quote, QuoteEvent, QuoteProjects, Supplier, Termsheet, TermsheetScheduleStep, ) 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) -> 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 "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": {}} if "category" in expand: cat_ids = {m.category_id for m in materials} rows = (await db.execute(select(MaterialCategory).where(MaterialCategory.id.in_(cat_ids)))).scalars().all() maps["categories"] = {c.id: c for c in rows} 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"] = category_brief(maps["categories"].get(m.category_id)) else: d["category_id"] = str(m.category_id) 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, ItemReceipt] = {} 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]] = {} 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} cat_ids = {m.category_id for m in materials} cats = (await db.execute(select(MaterialCategory).where(MaterialCategory.id.in_(cat_ids)))).scalars().all() data.categories = {c.id: c for c in cats} 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_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() data.receipts_by_decision = {r.item_decision_id: r for r in receipts} 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": category_brief(data.categories.get(m.category_id)), "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) if "decision" in expand: d["decision"] = decision_brief(decision) 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: r = data.receipts_by_decision.get(decision.id) if decision is not None else None d["receipt"] = receipt_brief(r) if r is not None else 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