from sqlalchemy import distinct, insert, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.models.imported_part import ImportedPart _FIELDS = frozenset( { "InventoryID", "Description", "TBXColour", "ProductMaxThreadDepthmm", "ProductMaxThreadDepthin", "ProductODmm", "ProductODin", "TBXThreadAttribute", } ) async def get(db: AsyncSession, inventory_id: str) -> ImportedPart | None: return await db.get(ImportedPart, inventory_id) async def list_all(db: AsyncSession, skip: int = 0, limit: int = 500) -> list[ImportedPart]: result = await db.execute( select(ImportedPart).order_by(ImportedPart.InventoryID).offset(skip).limit(limit) ) return list(result.scalars().all()) async def list_filtered( db: AsyncSession, description: str = "", skip: int = 0, limit: int = 500, ) -> list[ImportedPart]: q = select(ImportedPart) if description.strip(): q = q.where(ImportedPart.Description.ilike(f"%{description.strip()}%")) result = await db.execute(q.order_by(ImportedPart.InventoryID).offset(skip).limit(limit)) return list(result.scalars().all()) async def distinct_colours(db: AsyncSession) -> list[str]: result = await db.execute( select(distinct(ImportedPart.TBXColour)) .where(ImportedPart.TBXColour.isnot(None)) .order_by(ImportedPart.TBXColour) ) return [row[0] for row in result.all()] async def count(db: AsyncSession) -> int: from sqlalchemy import func result = await db.execute(select(func.count()).select_from(ImportedPart)) return result.scalar_one() async def count_filtered(db: AsyncSession, description: str = "") -> int: from sqlalchemy import func q = select(func.count()).select_from(ImportedPart) if description.strip(): q = q.where(ImportedPart.Description.ilike(f"%{description.strip()}%")) result = await db.execute(q) return result.scalar_one() async def delete(db: AsyncSession, part: ImportedPart) -> None: await db.delete(part) await db.commit() async def upsert_bulk(db: AsyncSession, records: list[dict]) -> tuple[int, int]: """Insert-or-update records keyed on InventoryID. Returns (created, updated).""" if not records: return 0, 0 clean = [ {k: str(v).strip() if v is not None else None for k, v in r.items() if k in _FIELDS} for r in records if r.get("InventoryID") ] if not clean: return 0, 0 incoming_ids = {r["InventoryID"] for r in clean} result = await db.execute( select(ImportedPart.InventoryID).where(ImportedPart.InventoryID.in_(incoming_ids)) ) existing_ids = {row[0] for row in result.all()} to_insert = [r for r in clean if r["InventoryID"] not in existing_ids] to_update = [r for r in clean if r["InventoryID"] in existing_ids] if to_insert: await db.execute(insert(ImportedPart), to_insert) for record in to_update: inv_id = record["InventoryID"] vals = {k: v for k, v in record.items() if k != "InventoryID"} if vals: await db.execute( update(ImportedPart).where(ImportedPart.InventoryID == inv_id).values(vals) ) await db.commit() return len(to_insert), len(to_update)