Merge pull request #24 from mlibrary/changing-items-filters

Changing items filters

aim/digifeeds/database/crud.py

@@ -24,36 +24,60 @@ def get_item(db: Session, barcode: str):
     return db.query(models.Item).filter(models.Item.barcode == barcode).first()
 
 
-def get_items_total(db: Session, in_zephir: bool | None):
-    query = get_items_query(db=db, in_zephir=in_zephir)
+def get_items_total(db: Session, filter: schemas.ItemFilters = None):
+    query = get_items_query(db=db, filter=filter)
     return query.count()
 
 
-def get_items(db: Session, in_zephir: bool | None, limit: int, offset: int):
+def get_items(
+    db: Session,
+    limit: int,
+    offset: int,
+    filter: schemas.ItemFilters = None,
+):
     """
     Get Digifeed items from the database
 
     Args:
         db (sqlalchemy.orm.Session): Digifeeds database session
-        in_zephir (bool | None): Whether or not the items are in zephir
+        filter (schemas.ItemFilters | None): filter to apply to the list of items.
 
     Returns:
         aim.digifeeds.database.models.Item: Item object
     """
-    query = get_items_query(db=db, in_zephir=in_zephir)
+    query = get_items_query(db=db, filter=filter)
     return query.offset(offset).limit(limit).all()
 
 
-def get_items_query(db: Session, in_zephir: bool | None):
+def get_items_query(db: Session, filter: schemas.ItemFilters = None):
     query = db.query(models.Item)
-    if in_zephir is True:
+
+    if filter == "in_zephir":
         query = query.filter(
             models.Item.statuses.any(models.ItemStatus.status_name == "in_zephir")
         )
-    elif in_zephir is False:
+    elif filter == "not_in_zephir":
         query = query.filter(
             ~models.Item.statuses.any(models.ItemStatus.status_name == "in_zephir")
         )
+    elif filter == "pending_deletion":
+        query = query.filter(
+            models.Item.statuses.any(
+                models.ItemStatus.status_name == "pending_deletion"
+            )
+        )
+    elif filter == "not_pending_deletion":
+        query = query.filter(
+            ~models.Item.statuses.any(
+                models.ItemStatus.status_name == "pending_deletion"
+            )
+        )
+    elif filter == "not_found_in_alma":
+        query = query.filter(
+            models.Item.statuses.any(
+                models.ItemStatus.status_name == "not_found_in_alma"
+            )
+        )
     return query
 
 

aim/digifeeds/database/main.py

@@ -43,23 +43,24 @@ def get_db():  # pragma: no cover
 def get_items(
     offset: int = Query(0, ge=0, description="Requested offset from the list of pages"),
     limit: int = Query(50, ge=1, description="Requested number of items per page"),
-    in_zephir: bool | None = Query(
-        None, description="Filter for items that do or do not have metadata in Zephir"
+    filter: schemas.ItemFilters = Query(
+        None, description="Filters on the items in the database"
     ),
     db: Session = Depends(get_db),
 ) -> schemas.PageOfItems:  # list[schemas.Item]:
     """
     Get the digifeeds items.
 
-    These items can be filtered by whether or not their metadata is in Zephir or
-    all of them can be fetched.
+    These items can be filtered by whether or not their metadata is in Zephir,
+    whether or not they are pending deletion, if they are not in alma, or all of
+    them can be fetched.
     """
 
-    db_items = crud.get_items(in_zephir=in_zephir, db=db, offset=offset, limit=limit)
+    db_items = crud.get_items(filter=filter, db=db, offset=offset, limit=limit)
     return {
         "limit": limit,
         "offset": offset,
-        "total": crud.get_items_total(in_zephir=in_zephir, db=db),
+        "total": crud.get_items_total(filter=filter, db=db),
         "items": db_items,
     }
 

aim/digifeeds/database/schemas.py

@@ -2,6 +2,7 @@
 
 from pydantic import BaseModel, Field, ConfigDict
 from datetime import datetime
+from enum import Enum
 
 
 class ItemStatus(BaseModel):
@@ -47,6 +48,14 @@ class PageOfItems(BaseModel):
     total: int = 1
 
 
+class ItemFilters(str, Enum):
+    in_zephir = "in_zephir"
+    not_in_zephir = "not_in_zephir"
+    pending_deletion = "pending_deletion"
+    not_pending_deletion = "not_pending_deletion"
+    not_found_in_alma = "not_found_in_alma"
+
+
 class ItemCreate(ItemBase):
     pass
 

aim/services.py

@@ -11,7 +11,7 @@
     structlog.processors.add_log_level,
 ]
 
-if sys.stderr.isatty():
+if sys.stderr.isatty():  # pragma: no cover
     # Pretty printing when we run in a terminal session.
     # Automatically prints pretty tracebacks when "rich" is installed
 

tests/digifeeds/database/test_crud.py

@@ -26,8 +26,8 @@ def test_get_items_and_total_any(self, db_session):
         item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
         status = get_status(db=db_session, name="in_zephir")
         add_item_status(db=db_session, item=item1, status=status)
-        items = get_items(db=db_session, in_zephir=None, limit=2, offset=0)
-        count = get_items_total(db=db_session, in_zephir=None)
+        items = get_items(db=db_session, filter=None, limit=2, offset=0)
+        count = get_items_total(db=db_session, filter=None)
         db_session.refresh(item1)
         db_session.refresh(item2)
         assert (items[0]) == item1
@@ -39,8 +39,8 @@ def test_get_items_and_total_in_zephir(self, db_session):
         item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
         status = get_status(db=db_session, name="in_zephir")
         add_item_status(db=db_session, item=item1, status=status)
-        items = get_items(db=db_session, in_zephir=True, limit=2, offset=0)
-        count = get_items_total(db=db_session, in_zephir=True)
+        items = get_items(db=db_session, filter="in_zephir", limit=2, offset=0)
+        count = get_items_total(db=db_session, filter="in_zephir")
         db_session.refresh(item1)
         db_session.refresh(item2)
         assert (len(items)) == 1
@@ -52,14 +52,55 @@ def test_get_items_and_total_not_in_zephir(self, db_session):
         item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
         status = get_status(db=db_session, name="in_zephir")
         add_item_status(db=db_session, item=item1, status=status)
-        items = get_items(db=db_session, in_zephir=False, limit=2, offset=0)
-        count = get_items_total(db=db_session, in_zephir=False)
+        items = get_items(db=db_session, filter="not_in_zephir", limit=2, offset=0)
+        count = get_items_total(db=db_session, filter="not_in_zephir")
         db_session.refresh(item1)
         db_session.refresh(item2)
         assert (len(items)) == 1
         assert (items[0]) == item2
         assert count == 1
 
+    def test_get_items_and_total_pending_deletion(self, db_session):
+        item1 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
+        status = get_status(db=db_session, name="pending_deletion")
+        add_item_status(db=db_session, item=item1, status=status)
+        items = get_items(db=db_session, filter="pending_deletion", limit=2, offset=0)
+        count = get_items_total(db=db_session, filter="pending_deletion")
+        db_session.refresh(item1)
+        db_session.refresh(item2)
+        assert (len(items)) == 1
+        assert (items[0]) == item1
+        assert count == 1
+
+    def test_get_items_and_total_not_pending_deletion(self, db_session):
+        item1 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
+        status = get_status(db=db_session, name="pending_deletion")
+        add_item_status(db=db_session, item=item1, status=status)
+        items = get_items(
+            db=db_session, filter="not_pending_deletion", limit=2, offset=0
+        )
+        count = get_items_total(db=db_session, filter="pending_deletion")
+        db_session.refresh(item1)
+        db_session.refresh(item2)
+        assert (len(items)) == 1
+        assert (items[0]) == item2
+        assert count == 1
+
+    def test_get_items_and_total_not_found_in_alma(self, db_session):
+        item1 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
+        status = get_status(db=db_session, name="not_found_in_alma")
+        add_item_status(db=db_session, item=item1, status=status)
+        items = get_items(db=db_session, filter="not_found_in_alma", limit=2, offset=0)
+        count = get_items_total(db=db_session, filter="not_found_in_alma")
+        db_session.refresh(item1)
+        db_session.refresh(item2)
+        assert (len(items)) == 1
+        assert (items[0]) == item1
+        assert count == 1
+
     def test_get_status_that_exists(self, db_session):
         status = get_status(db=db_session, name="in_zephir")
         assert (status.name) == "in_zephir"

tests/digifeeds/database/test_main.py

@@ -33,7 +33,7 @@ def test_get_items(client, valid_item, valid_in_zephir_item, db_session):
 def test_get_items_with_in_zephir_true(client, valid_item, valid_in_zephir_item):
     valid_item
     valid_in_zephir_item
-    response = client.get("/items", params={"in_zephir": True})
+    response = client.get("/items", params={"filter": "in_zephir"})
     assert response.status_code == 200, response.text
     assert len(response.json()["items"]) == 1
     assert response.json()["items"][0]["barcode"] == valid_in_zephir_item.barcode
@@ -42,7 +42,7 @@ def test_get_items_with_in_zephir_true(client, valid_item, valid_in_zephir_item)
 def test_get_items_with_in_zephir_false(client, valid_item, valid_in_zephir_item):
     valid_item
     valid_in_zephir_item
-    response = client.get("/items", params={"in_zephir": False})
+    response = client.get("/items", params={"filter": "not_in_zephir"})
     assert response.status_code == 200, response.text
     assert len(response.json()["items"]) == 1
     assert response.json()["items"][0]["barcode"] == valid_item.barcode