151 lines
4.4 KiB
Python
151 lines
4.4 KiB
Python
from sqlalchemy import distinct, insert, or_, 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",
|
|
}
|
|
)
|
|
|
|
# Sortable columns for the admin list, keyed by the query-string "sort" value.
|
|
SORT_COLUMNS = {
|
|
"InventoryID": ImportedPart.InventoryID,
|
|
"Description": ImportedPart.Description,
|
|
"TBXColour": ImportedPart.TBXColour,
|
|
"ProductMaxThreadDepthmm": ImportedPart.ProductMaxThreadDepthmm,
|
|
"ProductMaxThreadDepthin": ImportedPart.ProductMaxThreadDepthin,
|
|
"ProductODmm": ImportedPart.ProductODmm,
|
|
"ProductODin": ImportedPart.ProductODin,
|
|
"TBXThreadAttribute": ImportedPart.TBXThreadAttribute,
|
|
}
|
|
|
|
|
|
def _apply_sort(q, sort: str, direction: str):
|
|
column = SORT_COLUMNS.get(sort, ImportedPart.InventoryID)
|
|
order = column.desc() if direction == "desc" else column.asc()
|
|
return q.order_by(order, ImportedPart.InventoryID)
|
|
|
|
|
|
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 = "",
|
|
sort: str = "InventoryID",
|
|
direction: str = "asc",
|
|
skip: int = 0,
|
|
limit: int = 500,
|
|
) -> list[ImportedPart]:
|
|
q = select(ImportedPart)
|
|
if description.strip():
|
|
term = f"%{description.strip()}%"
|
|
q = q.where(
|
|
or_(ImportedPart.Description.ilike(term), ImportedPart.InventoryID.ilike(term))
|
|
)
|
|
q = _apply_sort(q, sort, direction)
|
|
result = await db.execute(q.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():
|
|
term = f"%{description.strip()}%"
|
|
q = q.where(
|
|
or_(ImportedPart.Description.ilike(term), ImportedPart.InventoryID.ilike(term))
|
|
)
|
|
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)
|