Merge pull request #29 from mlibrary/tidy-switch-and-named-tuple

Tidy switch and named tuple

aim/digifeeds/database/crud.py

@@ -14,6 +14,10 @@ class NotFoundError(Exception):
     pass
 
 
+class AlreadyExistsError(Exception):
+    pass
+
+
 def get_item(db: Session, barcode: str):
     """
     Get item from the database
@@ -69,33 +73,33 @@ def get_items(
 
 def get_items_statement(filter: schemas.ItemFilters = None):
     stmnt = select(models.Item)
-
-    if filter == "in_zephir":
-        stmnt = stmnt.where(
-            models.Item.statuses.any(models.ItemStatus.status_name == "in_zephir")
-        )
-    elif filter == "not_in_zephir":
-        stmnt = stmnt.where(
-            ~models.Item.statuses.any(models.ItemStatus.status_name == "in_zephir")
-        )
-    elif filter == "pending_deletion":
-        stmnt = stmnt.where(
-            models.Item.statuses.any(
-                models.ItemStatus.status_name == "pending_deletion"
+    match filter:
+        case "in_zephir":
+            stmnt = stmnt.where(
+                models.Item.statuses.any(models.ItemStatus.status_name == "in_zephir")
             )
-        )
-    elif filter == "not_pending_deletion":
-        stmnt = stmnt.where(
-            ~models.Item.statuses.any(
-                models.ItemStatus.status_name == "pending_deletion"
+        case "not_in_zephir":
+            stmnt = stmnt.where(
+                ~models.Item.statuses.any(models.ItemStatus.status_name == "in_zephir")
             )
-        )
-    elif filter == "not_found_in_alma":
-        stmnt = stmnt.where(
-            models.Item.statuses.any(
-                models.ItemStatus.status_name == "not_found_in_alma"
+        case "pending_deletion":
+            stmnt = stmnt.where(
+                models.Item.statuses.any(
+                    models.ItemStatus.status_name == "pending_deletion"
+                )
+            )
+        case "not_pending_deletion":
+            stmnt = stmnt.where(
+                ~models.Item.statuses.any(
+                    models.ItemStatus.status_name == "pending_deletion"
+                )
+            )
+        case "not_found_in_alma":
+            stmnt = stmnt.where(
+                models.Item.statuses.any(
+                    models.ItemStatus.status_name == "not_found_in_alma"
+                )
             )
-        )
     return stmnt
 
 
@@ -110,6 +114,13 @@ def add_item(db: Session, item: schemas.ItemCreate):
         aim.digifeeds.database.models.Item: Item object
     """
     db_item = models.Item(barcode=item.barcode)
+
+    stmnt = select(models.Item).filter_by(barcode=item.barcode)
+    already_in_db_item = db.scalars(stmnt).first()
+
+    if already_in_db_item is not None:
+        raise AlreadyExistsError()
+
     db.add(db_item)
     db.commit()
     db.refresh(db_item)

aim/digifeeds/database/main.py

@@ -7,7 +7,7 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
 from aim.digifeeds.database import crud, schemas
-from aim.digifeeds.database.crud import NotFoundError
+from aim.digifeeds.database.crud import NotFoundError, AlreadyExistsError
 from aim.services import S
 
 # This is here so SessionLocal won't have a problem in tests in github
@@ -117,12 +117,11 @@ def create_item(
 
     item = schemas.ItemCreate(barcode=barcode)
     try:
-        crud.get_item(barcode=item.barcode, db=db)
-    except NotFoundError:
         db_item = crud.add_item(item=item, db=db)
-        return db_item
-    else:
+    except AlreadyExistsError:
         raise HTTPException(status_code=400, detail="Item already exists")
+    else:
+        return db_item
 
 
 desc_put_404 = """

aim/services.py

@@ -1,4 +1,5 @@
-from typing import NamedTuple
+# from typing import NamedTuple
+from dataclasses import dataclass
 import os
 import sqlalchemy as sa
 import structlog
@@ -32,7 +33,8 @@
 structlog.configure(processors)
 
 
-class Services(NamedTuple):
+@dataclass(frozen=True)
+class Services:
     """
     Global Configuration Services
     """

docs/conf.py

@@ -21,7 +21,6 @@
     "sphinx.ext.autodoc",
     "myst_parser",
     "sphinxcontrib.mermaid",
-    "sphinx_toolbox.more_autodoc.autonamedtuple",
 ]
 autosummary_generate = True