Adding pagination

app/crud/augmented_part.py

@@ -1,4 +1,4 @@
-from sqlalchemy import select
+from sqlalchemy import distinct, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.associations import augmented_part_imported_parts
@@ -52,10 +52,39 @@ async def list_filtered(
     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,
+) -> int:
+    from sqlalchemy import func
+    q = select(func.count()).select_from(AugmentedPart)
+    if description.strip():
+        q = q.where(AugmentedPart.description.ilike(f"%{description.strip()}%"))
+    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)
+    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)
@@ -69,10 +98,48 @@ async def get_unlinked_imported_parts_filtered(
         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))
+    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,
     *,

app/crud/imported_part.py

@@ -51,6 +51,15 @@ async def count(db: AsyncSession) -> int:
     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()