Serverside sort and limit for vendor and filaments

spoolman/database/filament.py

@@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import contains_eager, joinedload
 
 from spoolman.database import models, vendor
+from spoolman.database.utils import SortOrder, parse_nested_field
 from spoolman.exceptions import ItemDeleteError, ItemNotFoundError
 
 
@@ -65,7 +66,7 @@ async def get_by_id(db: AsyncSession, filament_id: int) -> models.Filament:
     return filament
 
 
-async def find(
+async def find(  # noqa: C901
     *,
     db: AsyncSession,
     vendor_name: Optional[str] = None,
@@ -73,8 +74,15 @@ async def find(
     name: Optional[str] = None,
     material: Optional[str] = None,
     article_number: Optional[str] = None,
+    sort_by: Optional[dict[str, SortOrder]] = None,
+    limit: Optional[int] = None,
+    offset: int = 0,
 ) -> list[models.Filament]:
-    """Find a list of filament objects by search criteria."""
+    """Find a list of filament objects by search criteria.
+
+    Sort by a field by passing a dict with the field name as key and the sort order as value.
+    The field name can contain nested fields, e.g. vendor.name.
+    """
     stmt = (
         select(models.Filament)
         .options(contains_eager(models.Filament.vendor))
@@ -91,6 +99,17 @@ async def find(
     if article_number is not None:
         stmt = stmt.where(models.Filament.article_number.ilike(f"%{article_number}%"))
 
+    if sort_by is not None:
+        for fieldstr, order in sort_by.items():
+            field = parse_nested_field(models.Filament, fieldstr)
+            if order == SortOrder.ASC:
+                stmt = stmt.order_by(field.asc())
+            elif order == SortOrder.DESC:
+                stmt = stmt.order_by(field.desc())
+
+    if limit is not None:
+        stmt = stmt.offset(offset).limit(limit)
+
     rows = await db.execute(stmt)
     return list(rows.scalars().all())
 

spoolman/database/spool.py

@@ -10,7 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import contains_eager, joinedload
 
 from spoolman.database import filament, models
-from spoolman.database.utils import SortOrder
+from spoolman.database.utils import SortOrder, parse_nested_field
 from spoolman.exceptions import ItemCreateError, ItemNotFoundError
 from spoolman.math import weight_from_length
 
@@ -76,28 +76,6 @@ async def get_by_id(db: AsyncSession, spool_id: int) -> models.Spool:
     return spool
 
 
-def parse_nested_field(base_obj: type[models.Base], field: str) -> sqlalchemy.Column:
-    """Parse a nested field string into a sqlalchemy field object."""
-    fields = field.split(".")
-    if not hasattr(base_obj, fields[0]):
-        raise ValueError(f"Invalid field name '{field}'")
-
-    if fields[0] == "filament" and len(fields) == 1:
-        raise ValueError("No field specified for filament")
-    if fields[0] == "filament":
-        return parse_nested_field(models.Filament, ".".join(fields[1:]))
-
-    if fields[0] == "vendor" and len(fields) == 1:
-        raise ValueError("No field specified for vendor")
-    if fields[0] == "vendor":
-        return parse_nested_field(models.Vendor, ".".join(fields[1:]))
-
-    if len(fields) > 1:
-        raise ValueError(f"Field '{fields[0]}' does not have any nested fields")
-
-    return getattr(base_obj, fields[0])
-
-
 async def find(  # noqa: C901, PLR0912
     *,
     db: AsyncSession,

spoolman/database/utils.py

@@ -2,7 +2,33 @@
 
 from enum import Enum
 
+import sqlalchemy
+
+from spoolman.database import models
+
 
 class SortOrder(Enum):
     ASC = 1
     DESC = 2
+
+
+def parse_nested_field(base_obj: type[models.Base], field: str) -> sqlalchemy.Column:
+    """Parse a nested field string into a sqlalchemy field object."""
+    fields = field.split(".")
+    if not hasattr(base_obj, fields[0]):
+        raise ValueError(f"Invalid field name '{field}'")
+
+    if fields[0] == "filament" and len(fields) == 1:
+        raise ValueError("No field specified for filament")
+    if fields[0] == "filament":
+        return parse_nested_field(models.Filament, ".".join(fields[1:]))
+
+    if fields[0] == "vendor" and len(fields) == 1:
+        raise ValueError("No field specified for vendor")
+    if fields[0] == "vendor":
+        return parse_nested_field(models.Vendor, ".".join(fields[1:]))
+
+    if len(fields) > 1:
+        raise ValueError(f"Field '{fields[0]}' does not have any nested fields")
+
+    return getattr(base_obj, fields[0])

spoolman/database/vendor.py

@@ -6,6 +6,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from spoolman.database import models
+from spoolman.database.utils import SortOrder
 from spoolman.exceptions import ItemNotFoundError
 
 
@@ -37,12 +38,26 @@ async def find(
     *,
     db: AsyncSession,
     name: Optional[str] = None,
+    sort_by: Optional[dict[str, SortOrder]] = None,
+    limit: Optional[int] = None,
+    offset: int = 0,
 ) -> list[models.Vendor]:
     """Find a list of vendor objects by search criteria."""
     stmt = select(models.Vendor)
     if name is not None:
         stmt = stmt.where(models.Vendor.name.ilike(f"%{name}%"))
 
+    if sort_by is not None:
+        for fieldstr, order in sort_by.items():
+            field = getattr(models.Vendor, fieldstr)
+            if order == SortOrder.ASC:
+                stmt = stmt.order_by(field.asc())
+            elif order == SortOrder.DESC:
+                stmt = stmt.order_by(field.desc())
+
+    if limit is not None:
+        stmt = stmt.offset(offset).limit(limit)
+
     rows = await db.execute(stmt)
     return list(rows.scalars().all())