290 lines
9.3 KiB
Python
290 lines
9.3 KiB
Python
from sqlalchemy import distinct, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.associations import augmented_part_imported_parts
|
|
from app.models.augmented_part import AugmentedPart
|
|
from app.models.imported_part import ImportedPart
|
|
from app.models.manufacturer import Manufacturer
|
|
from app.models.part_colour import PartColour
|
|
from app.models.part_type import PartType
|
|
from app.models.scope_model import ScopeModel
|
|
from app.models.sub_part_type import SubPartType
|
|
|
|
_RELATIONSHIPS = [
|
|
"colour",
|
|
"manufacturer",
|
|
"scope_model",
|
|
"part_type",
|
|
"sub_part_type",
|
|
"notes",
|
|
]
|
|
|
|
# Sortable columns for the admin list, keyed by the query-string "sort" value.
|
|
# Columns backed by a related table's name require an outer join (the FK may be
|
|
# nullable, e.g. scope_model_id) so unrelated rows still appear, ordered last.
|
|
SORT_COLUMNS = {
|
|
"id": (AugmentedPart.id, None),
|
|
"sku": (AugmentedPart.sku, None),
|
|
"description": (AugmentedPart.description, None),
|
|
"colour": (PartColour.name, (PartColour, AugmentedPart.colour_id == PartColour.id)),
|
|
"manufacturer": (
|
|
Manufacturer.name,
|
|
(Manufacturer, AugmentedPart.manufacturer_id == Manufacturer.id),
|
|
),
|
|
"scope_model": (
|
|
ScopeModel.name,
|
|
(ScopeModel, AugmentedPart.scope_model_id == ScopeModel.id),
|
|
),
|
|
"part_type": (PartType.name, (PartType, AugmentedPart.part_type_id == PartType.id)),
|
|
"sub_part_type": (
|
|
SubPartType.name,
|
|
(SubPartType, AugmentedPart.sub_part_type_id == SubPartType.id),
|
|
),
|
|
}
|
|
|
|
|
|
def _apply_sort(q, sort: str, direction: str):
|
|
column, join = SORT_COLUMNS.get(sort, SORT_COLUMNS["description"])
|
|
if join is not None:
|
|
table, cond = join
|
|
q = q.outerjoin(table, cond)
|
|
order = column.desc() if direction == "desc" else column.asc()
|
|
# AugmentedPart.id as a tiebreaker keeps paginated ordering stable.
|
|
return q.order_by(order, AugmentedPart.id)
|
|
|
|
|
|
async def get(db: AsyncSession, augmented_part_id: int) -> AugmentedPart | None:
|
|
return await db.get(AugmentedPart, augmented_part_id)
|
|
|
|
|
|
async def get_by_sku(db: AsyncSession, sku: str) -> AugmentedPart | None:
|
|
result = await db.execute(
|
|
select(AugmentedPart).where(AugmentedPart.sku == sku).order_by(AugmentedPart.id)
|
|
)
|
|
return result.scalars().first()
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
include_archived: bool = False,
|
|
) -> list[AugmentedPart]:
|
|
q = select(AugmentedPart).order_by(AugmentedPart.description)
|
|
if not include_archived:
|
|
q = q.where(AugmentedPart.archived.is_(False))
|
|
result = await db.execute(q.offset(skip).limit(limit))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_many(db: AsyncSession, ids: list[int]) -> list[AugmentedPart]:
|
|
"""Fetch augmented parts by id in one query (selectin relationships loaded)."""
|
|
if not ids:
|
|
return []
|
|
result = await db.execute(select(AugmentedPart).where(AugmentedPart.id.in_(ids)))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_filtered(
|
|
db: AsyncSession,
|
|
description: str = "",
|
|
colour_id: int | None = None,
|
|
manufacturer_id: int | None = None,
|
|
scope_model_id: int | None = None,
|
|
part_type_id: int | None = None,
|
|
sub_part_type_id: int | None = None,
|
|
include_archived: bool = False,
|
|
on_website: bool = False,
|
|
sort: str = "description",
|
|
direction: str = "asc",
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> list[AugmentedPart]:
|
|
q = select(AugmentedPart)
|
|
if description.strip():
|
|
term = f"%{description.strip()}%"
|
|
q = q.where(
|
|
or_(AugmentedPart.description.ilike(term), AugmentedPart.sku.ilike(term))
|
|
)
|
|
if colour_id is not None:
|
|
q = q.where(AugmentedPart.colour_id == colour_id)
|
|
if manufacturer_id is not None:
|
|
q = q.where(AugmentedPart.manufacturer_id == manufacturer_id)
|
|
if scope_model_id is not None:
|
|
q = q.where(AugmentedPart.scope_model_id == scope_model_id)
|
|
if part_type_id is not None:
|
|
q = q.where(AugmentedPart.part_type_id == part_type_id)
|
|
if sub_part_type_id is not None:
|
|
q = q.where(AugmentedPart.sub_part_type_id == sub_part_type_id)
|
|
if not include_archived:
|
|
q = q.where(AugmentedPart.archived.is_(False))
|
|
if on_website:
|
|
q = q.where(AugmentedPart.on_website.is_(True))
|
|
q = _apply_sort(q, sort, direction)
|
|
result = await db.execute(q.offset(skip).limit(limit))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def count_filtered(
|
|
db: AsyncSession,
|
|
description: str = "",
|
|
colour_id: int | None = None,
|
|
manufacturer_id: int | None = None,
|
|
scope_model_id: int | None = None,
|
|
part_type_id: int | None = None,
|
|
sub_part_type_id: int | None = None,
|
|
include_archived: bool = False,
|
|
on_website: bool = False,
|
|
) -> int:
|
|
from sqlalchemy import func
|
|
|
|
q = select(func.count()).select_from(AugmentedPart)
|
|
if description.strip():
|
|
term = f"%{description.strip()}%"
|
|
q = q.where(
|
|
or_(AugmentedPart.description.ilike(term), AugmentedPart.sku.ilike(term))
|
|
)
|
|
if colour_id is not None:
|
|
q = q.where(AugmentedPart.colour_id == colour_id)
|
|
if manufacturer_id is not None:
|
|
q = q.where(AugmentedPart.manufacturer_id == manufacturer_id)
|
|
if scope_model_id is not None:
|
|
q = q.where(AugmentedPart.scope_model_id == scope_model_id)
|
|
if part_type_id is not None:
|
|
q = q.where(AugmentedPart.part_type_id == part_type_id)
|
|
if sub_part_type_id is not None:
|
|
q = q.where(AugmentedPart.sub_part_type_id == sub_part_type_id)
|
|
if not include_archived:
|
|
q = q.where(AugmentedPart.archived.is_(False))
|
|
if on_website:
|
|
q = q.where(AugmentedPart.on_website.is_(True))
|
|
result = await db.execute(q)
|
|
return result.scalar_one()
|
|
|
|
|
|
async def get_unlinked_imported_parts_filtered(
|
|
db: AsyncSession,
|
|
description: str = "",
|
|
colour: str = "",
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
) -> list[ImportedPart]:
|
|
q = select(ImportedPart).where(
|
|
ImportedPart.InventoryID.not_in(
|
|
select(augmented_part_imported_parts.c.imported_part_id)
|
|
)
|
|
)
|
|
if description.strip():
|
|
q = q.where(ImportedPart.Description.ilike(f"%{description.strip()}%"))
|
|
if colour.strip():
|
|
q = q.where(ImportedPart.TBXColour.ilike(f"%{colour.strip()}%"))
|
|
result = await db.execute(
|
|
q.order_by(ImportedPart.InventoryID).offset(skip).limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def count_unlinked_filtered(
|
|
db: AsyncSession,
|
|
description: str = "",
|
|
colour: str = "",
|
|
) -> int:
|
|
from sqlalchemy import func
|
|
|
|
q = (
|
|
select(func.count())
|
|
.select_from(ImportedPart)
|
|
.where(
|
|
ImportedPart.InventoryID.not_in(
|
|
select(augmented_part_imported_parts.c.imported_part_id)
|
|
)
|
|
)
|
|
)
|
|
if description.strip():
|
|
q = q.where(ImportedPart.Description.ilike(f"%{description.strip()}%"))
|
|
if colour.strip():
|
|
q = q.where(ImportedPart.TBXColour.ilike(f"%{colour.strip()}%"))
|
|
result = await db.execute(q)
|
|
return result.scalar_one()
|
|
|
|
|
|
async def distinct_unlinked_colours(db: AsyncSession) -> list[str]:
|
|
q = (
|
|
select(distinct(ImportedPart.TBXColour))
|
|
.where(
|
|
ImportedPart.TBXColour.isnot(None),
|
|
ImportedPart.InventoryID.not_in(
|
|
select(augmented_part_imported_parts.c.imported_part_id)
|
|
),
|
|
)
|
|
.order_by(ImportedPart.TBXColour)
|
|
)
|
|
result = await db.execute(q)
|
|
return [row[0] for row in result.all()]
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
sku: str | None,
|
|
description: str,
|
|
colour_id: int,
|
|
manufacturer_id: int,
|
|
scope_model_id: int | None,
|
|
part_type_id: int,
|
|
sub_part_type_id: int | None,
|
|
on_website: bool = False,
|
|
preferred_part: bool = False,
|
|
) -> AugmentedPart:
|
|
ap = AugmentedPart(
|
|
sku=sku,
|
|
description=description,
|
|
colour_id=colour_id,
|
|
manufacturer_id=manufacturer_id,
|
|
scope_model_id=scope_model_id,
|
|
part_type_id=part_type_id,
|
|
sub_part_type_id=sub_part_type_id,
|
|
on_website=on_website,
|
|
preferred_part=preferred_part,
|
|
)
|
|
db.add(ap)
|
|
await db.commit()
|
|
await db.refresh(ap, attribute_names=_RELATIONSHIPS)
|
|
return ap
|
|
|
|
|
|
async def update(db: AsyncSession, ap: AugmentedPart, data: dict) -> AugmentedPart:
|
|
for field, value in data.items():
|
|
setattr(ap, field, value)
|
|
await db.commit()
|
|
await db.refresh(ap, attribute_names=_RELATIONSHIPS)
|
|
return ap
|
|
|
|
|
|
async def get_unlinked_imported_parts(db: AsyncSession) -> list[ImportedPart]:
|
|
"""ImportedParts that have no corresponding AugmentedPart link."""
|
|
result = await db.execute(
|
|
select(ImportedPart)
|
|
.where(
|
|
ImportedPart.InventoryID.not_in(
|
|
select(augmented_part_imported_parts.c.imported_part_id)
|
|
)
|
|
)
|
|
.order_by(ImportedPart.InventoryID)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def set_archived(
|
|
db: AsyncSession, ap: AugmentedPart, archived: bool
|
|
) -> AugmentedPart:
|
|
ap.archived = archived
|
|
await db.commit()
|
|
await db.refresh(ap, attribute_names=_RELATIONSHIPS)
|
|
return ap
|
|
|
|
|
|
async def delete(db: AsyncSession, ap: AugmentedPart) -> None:
|
|
await db.delete(ap)
|
|
await db.commit()
|