from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select from sqlalchemy.orm import selectinload from typing import List, Optional from uuid import UUID from app.core.database import get_db from app.models.bom_models import BomMaster, BomDetail, ProductMaster from app.core.security import require_permission from app.models.core_models import LoginUsers from app.models.rm_models import RmMaster, MaterialTypeMaster from . import schemas router = APIRouter() # ─── PRODUCT ENDPOINTS ──────────────────────────────────────────────────────── @router.get("/products", response_model=List[schemas.ProductResponse]) async def list_products(db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'view'))): result = await db.execute(select(ProductMaster).order_by(ProductMaster.name)) return result.scalars().all() @router.get("/products/{product_id}", response_model=schemas.ProductResponse) async def get_product(product_id: UUID, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'view'))): product = await db.get(ProductMaster, product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") return product @router.post("/products", response_model=schemas.ProductResponse, status_code=status.HTTP_201_CREATED) async def create_product(payload: schemas.ProductCreate, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'create'))): if payload.product_code: stmt = select(ProductMaster).where(ProductMaster.product_code == payload.product_code) existing = await db.execute(stmt) if existing.scalars().first(): raise HTTPException(status_code=400, detail="Product code already exists") product = ProductMaster(**payload.model_dump()) db.add(product) await db.commit() await db.refresh(product) return product @router.put("/products/{product_id}", response_model=schemas.ProductResponse) async def update_product(product_id: UUID, payload: schemas.ProductUpdate, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'edit'))): product = await db.get(ProductMaster, product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") update_data = payload.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(product, key, value) await db.commit() await db.refresh(product) return product # ─── BOM LIST (LEAN - no detail lines) ─────────────────────────────────────── @router.get("/", response_model=List[schemas.BomListResponse]) async def list_boms( skip: int = Query(0, ge=0), limit: int = Query(10000, ge=1, le=100000), search: Optional[str] = Query(None), db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'view')), ): """ Fast lean list: returns BOM headers + product name + component count only. Does NOT load 33k detail rows - those are loaded on the individual detail page. """ count_sq = ( select(BomDetail.bom_id, func.count(BomDetail.bom_detail_id).label("component_count")) .group_by(BomDetail.bom_id) .subquery() ) stmt = ( select( BomMaster.bom_id, BomMaster.product_id, BomMaster.bom_number, BomMaster.description, BomMaster.is_active, BomMaster.created_at, BomMaster.updated_at, ProductMaster.name.label("product_name"), ProductMaster.product_code.label("product_code"), func.coalesce(count_sq.c.component_count, 0).label("component_count"), ) .join(ProductMaster, BomMaster.product_id == ProductMaster.product_id, isouter=True) .outerjoin(count_sq, BomMaster.bom_id == count_sq.c.bom_id) .order_by(BomMaster.bom_number) ) if search: stmt = stmt.where( BomMaster.bom_number.ilike(f"%{search}%") | ProductMaster.name.ilike(f"%{search}%") ) stmt = stmt.offset(skip).limit(limit) rows = (await db.execute(stmt)).mappings().all() return [dict(r) for r in rows] # ─── SPECIAL FEATURE BOM MAPPING ───────────────────────────────────────────── from app.models.bom_models import SpecialFeatureBom from sqlalchemy.exc import IntegrityError from typing import Optional async def _feature_bom_rows(db: AsyncSession, *conditions) -> list[dict]: """Feature BOM lines with the part number, name and unit of each raw material.""" stmt = ( select( SpecialFeatureBom, RmMaster.name.label('rm_name'), RmMaster.part_no.label('rm_part_no'), RmMaster.unit_of_measurement.label('rm_uom'), ) .join(RmMaster, SpecialFeatureBom.rm_id == RmMaster.rm_id) .where(*conditions) .order_by(RmMaster.part_no) ) rows = (await db.execute(stmt)).all() return [ { "id": m.id, "option_id": m.option_id, "rm_id": m.rm_id, "quantity": m.quantity, "uom": m.uom, "created_at": m.created_at, "updated_at": m.updated_at, "rm_name": rm_name, "rm_part_no": rm_part_no, "rm_uom": rm_uom, } for m, rm_name, rm_part_no, rm_uom in rows ] @router.get("/feature-boms", response_model=List[schemas.SpecialFeatureBomResponse]) async def list_feature_boms( option_id: Optional[UUID] = Query(None), db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'view')) ): """Raw materials added by feature options. A feature option's parts apply to every model.""" conditions = [SpecialFeatureBom.option_id == option_id] if option_id else [] return await _feature_bom_rows(db, *conditions) @router.post("/feature-boms", response_model=schemas.SpecialFeatureBomResponse, status_code=status.HTTP_201_CREATED) async def create_feature_bom(payload: schemas.SpecialFeatureBomCreate, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'create'))): mapping = SpecialFeatureBom( option_id=payload.option_id, rm_id=payload.rm_id, quantity=payload.quantity, uom=payload.uom ) db.add(mapping) try: await db.commit() except IntegrityError: await db.rollback() raise HTTPException(status_code=400, detail="This material is already mapped to this feature option.") return (await _feature_bom_rows(db, SpecialFeatureBom.id == mapping.id))[0] @router.put("/feature-boms/{mapping_id}", response_model=schemas.SpecialFeatureBomResponse) async def update_feature_bom(mapping_id: UUID, payload: schemas.SpecialFeatureBomUpdate, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'edit'))): mapping = await db.get(SpecialFeatureBom, mapping_id) if not mapping: raise HTTPException(status_code=404, detail="Mapping not found") if payload.quantity is not None: mapping.quantity = payload.quantity if payload.uom is not None: mapping.uom = payload.uom try: await db.commit() except IntegrityError: await db.rollback() raise HTTPException(status_code=400, detail="This update causes a duplicate entry.") return (await _feature_bom_rows(db, SpecialFeatureBom.id == mapping_id))[0] @router.delete("/feature-boms/{mapping_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_feature_bom(mapping_id: UUID, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'delete'))): mapping = await db.get(SpecialFeatureBom, mapping_id) if not mapping: raise HTTPException(status_code=404, detail="Mapping not found") await db.delete(mapping) await db.commit() # ─── BOM DETAIL (full with components) ─────────────────────────────────────── @router.get("/{bom_id}", response_model=schemas.BomResponse) async def get_bom(bom_id: UUID, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'view'))): stmt = select(BomMaster).where(BomMaster.bom_id == bom_id).options( selectinload(BomMaster.product), selectinload(BomMaster.details) ) result = await db.execute(stmt) bom = result.scalars().first() if not bom: raise HTTPException(status_code=404, detail="BOM not found") rm_ids = [d.rm_id for d in bom.details] rm_map = {} if rm_ids: from app.models.rm_models import RmVendorMapping, VendorMaster # 1. Fetch RM and Material Type rm_stmt = ( select(RmMaster, MaterialTypeMaster.name.label("material_type_name")) .outerjoin(MaterialTypeMaster, RmMaster.material_type_id == MaterialTypeMaster.id) .where(RmMaster.rm_id.in_(rm_ids)) ) rm_res = await db.execute(rm_stmt) for rm, mat_type_name in rm_res.all(): rm_map[rm.rm_id] = {"rm": rm, "mat_type": mat_type_name, "suppliers": []} # 2. Fetch Suppliers vendor_stmt = ( select(RmVendorMapping.rm_id, VendorMaster.name) .join(VendorMaster, RmVendorMapping.vendor_id == VendorMaster.vendor_id) .where(RmVendorMapping.rm_id.in_(rm_ids)) .where(RmVendorMapping.is_active == True) ) vendor_res = await db.execute(vendor_stmt) for rm_id, vendor_name in vendor_res.all(): if rm_id in rm_map: rm_map[rm_id]["suppliers"].append(vendor_name) bom_dict = { "bom_id": bom.bom_id, "product_id": bom.product_id, "bom_number": bom.bom_number, "description": bom.description, "is_active": bom.is_active, "created_at": bom.created_at, "updated_at": bom.updated_at, "product_name": bom.product.name if bom.product else None, "product_code": bom.product.product_code if bom.product else None, "details": [] } for d in bom.details: rm_data = rm_map.get(d.rm_id) rm = rm_data["rm"] if rm_data else None mat_type = rm_data["mat_type"] if rm_data else None suppliers = ", ".join(rm_data["suppliers"]) if rm_data and rm_data["suppliers"] else "—" bom_dict["details"].append({ "bom_detail_id": d.bom_detail_id, "bom_id": d.bom_id, "rm_id": d.rm_id, "quantity": d.quantity, "uom": d.uom or (rm.unit_of_measurement if rm else None), "rm_name": rm.name if rm else "Unknown", "rm_part_no": rm.part_no if rm else None, "material_type": mat_type or "Unknown", "supplier_name": suppliers }) return bom_dict @router.post("/", response_model=schemas.BomResponse, status_code=status.HTTP_201_CREATED) async def create_bom(payload: schemas.BomCreate, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'create'))): product = await db.get(ProductMaster, payload.product_id) if not product: raise HTTPException(status_code=400, detail="Product not found") stmt = select(BomMaster).where(BomMaster.bom_number == payload.bom_number) existing = await db.execute(stmt) if existing.scalars().first(): raise HTTPException(status_code=400, detail="BOM number already exists") # Validate all rm_ids exist if payload.details: rm_ids = {d.rm_id for d in payload.details} rm_stmt = select(RmMaster.rm_id).where(RmMaster.rm_id.in_(rm_ids)) rm_result = await db.execute(rm_stmt) existing_rm_ids = set(rm_result.scalars().all()) missing_rm_ids = rm_ids - existing_rm_ids if missing_rm_ids: missing_str = ", ".join(str(rid) for rid in missing_rm_ids) raise HTTPException(status_code=400, detail=f"Invalid Raw Material ID(s): {missing_str}") # Validate no duplicate raw materials in the details rm_ids_list = [d.rm_id for d in payload.details] if len(rm_ids_list) != len(set(rm_ids_list)): raise HTTPException(status_code=400, detail="Duplicate raw materials are not allowed in the same BOM") if payload.is_active: from sqlalchemy import update deactivate_stmt = ( update(BomMaster) .where(BomMaster.product_id == payload.product_id, BomMaster.is_active == True) .values(is_active=False) ) await db.execute(deactivate_stmt) bom = BomMaster( product_id=payload.product_id, bom_number=payload.bom_number, description=payload.description, is_active=payload.is_active ) db.add(bom) await db.flush() for d in payload.details: detail = BomDetail( bom_id=bom.bom_id, rm_id=d.rm_id, quantity=d.quantity, uom=d.uom ) db.add(detail) await db.commit() return await get_bom(bom.bom_id, db) @router.put("/{bom_id}", response_model=schemas.BomResponse) async def update_bom(bom_id: UUID, payload: schemas.BomUpdate, db: AsyncSession = Depends(get_db), current_user: LoginUsers = Depends(require_permission('BOM', 'edit'))): stmt = select(BomMaster).where(BomMaster.bom_id == bom_id).options(selectinload(BomMaster.details)) result = await db.execute(stmt) bom = result.scalars().first() if not bom: raise HTTPException(status_code=404, detail="BOM not found") if payload.bom_number is not None and payload.bom_number != bom.bom_number: stmt_check = select(BomMaster).where(BomMaster.bom_number == payload.bom_number) existing = await db.execute(stmt_check) if existing.scalars().first(): raise HTTPException(status_code=400, detail="BOM number already exists") bom.bom_number = payload.bom_number if payload.description is not None: bom.description = payload.description if payload.is_active is not None: bom.is_active = payload.is_active if payload.is_active: from sqlalchemy import update deactivate_stmt = ( update(BomMaster) .where( BomMaster.product_id == bom.product_id, BomMaster.is_active == True, BomMaster.bom_id != bom_id ) .values(is_active=False) ) await db.execute(deactivate_stmt) if payload.details is not None: # Validate all rm_ids exist if payload.details: rm_ids = {d.rm_id for d in payload.details} rm_stmt = select(RmMaster.rm_id).where(RmMaster.rm_id.in_(rm_ids)) rm_result = await db.execute(rm_stmt) existing_rm_ids = set(rm_result.scalars().all()) missing_rm_ids = rm_ids - existing_rm_ids if missing_rm_ids: missing_str = ", ".join(str(rid) for rid in missing_rm_ids) raise HTTPException(status_code=400, detail=f"Invalid Raw Material ID(s): {missing_str}") # Validate no duplicate raw materials in the details rm_ids_list = [d.rm_id for d in payload.details] if len(rm_ids_list) != len(set(rm_ids_list)): raise HTTPException(status_code=400, detail="Duplicate raw materials are not allowed in the same BOM") for d in bom.details: await db.delete(d) bom.details.clear() for d in payload.details: detail = BomDetail( bom_id=bom.bom_id, rm_id=d.rm_id, quantity=d.quantity, uom=d.uom ) db.add(detail) bom.details.append(detail) await db.commit() return await get_bom(bom_id, db)