Merge pull request #14 from mlibrary/DWI-14-check-zephir

DWI-14 check zephir

aim/cli/digifeeds.py

@@ -1,10 +1,12 @@
 """Digifeeds CLI
 ====================
 """
+
 import typer
 from typing_extensions import Annotated
 from aim.digifeeds.add_to_db import add_to_db as add_to_digifeeds_db
 from aim.digifeeds.list_barcodes_in_bucket import list_barcodes_in_bucket
+from aim.digifeeds.check_zephir import check_zephir as check_zephir_for_barcode
 from aim.digifeeds.database import models, main
 import json
 import sys
@@ -40,6 +42,27 @@ def add_to_db(
         print("Item NOT added to digifeeds set")
 
 
+@app.command()
+def check_zephir(
+    barcode: Annotated[
+        str,
+        typer.Argument(
+            help="The barcode to check in zephir. It should NOT have mdp prefix. The barcode must already exist in the digifeeds database."
+        ),
+    ],
+):
+    """
+    Check if barcode has metadata in Zephir
+    """
+
+    print(f"Checking Zephir for {barcode}")
+    item = check_zephir_for_barcode(barcode)
+    if item:
+        print(f"{barcode} is in Zephir")
+    else:
+        print(f"{barcode} is NOT in Zephir")
+
+
 @app.command()
 def load_statuses():
     """

aim/digifeeds/check_zephir.py

@@ -0,0 +1,22 @@
+from aim.services import S
+from aim.digifeeds.db_client import DBClient
+from aim.digifeeds.item import Item
+import requests
+
+
+def check_zephir(barcode: str):
+    raw_item = DBClient().get_item(barcode)
+    if raw_item is None:
+        raise Exception("Item not found in database")
+
+    item = Item(raw_item)
+
+    if item.has_status("in_zephir"):
+        return item
+
+    response = requests.get(f"{S.zephir_bib_api_url}/mdp.{barcode}")
+    if response.status_code == 200:
+        DBClient().add_item_status(barcode=barcode, status="in_zephir")
+        return item
+    else:
+        return None

aim/services.py

@@ -41,6 +41,9 @@ class Services(NamedTuple):
     #: The url in the s3 bucket for the digifeeds process
     digifeeds_s3_input_path: str
 
+    #: The zephir item bib api
+    zephir_bib_api_url: str
+
 
 S = Services(
     mysql_database=sa.engine.URL.create(
@@ -60,8 +63,8 @@ class Services(NamedTuple):
     or "digifeeds_s3_access_key",
     digifeeds_s3_secret_access_key=os.getenv("DIGIFEEDS_S3_SECRET_ACCESS_KEY")
     or "digifeeds_s3_secret_access_key",
-    digifeeds_s3_bucket=os.getenv(
-        "DIGIFEEDS_S3_BUCKET") or "digifeeds_s3_bucket",
+    digifeeds_s3_bucket=os.getenv("DIGIFEEDS_S3_BUCKET") or "digifeeds_s3_bucket",
     digifeeds_s3_input_path=os.getenv("DIGIFEEDS_S3_INPUT_PATH")
     or "path_to_input_barcodes",
+    zephir_bib_api_url="http://zephir.cdlib.org/api/item",
 )

tests/cli/test_digifeeds.py

@@ -68,10 +68,8 @@ def test_add_to_db_where_item_is_not_in_alma(item_data, mocker):
 
 
 def test_load_statuses(mocker):
-    session_local_mock = mocker.patch(
-        "aim.digifeeds.database.main.SessionLocal")
-    load_statuse_mock = mocker.patch(
-        "aim.digifeeds.database.models.load_statuses")
+    session_local_mock = mocker.patch("aim.digifeeds.database.main.SessionLocal")
+    load_statuse_mock = mocker.patch("aim.digifeeds.database.models.load_statuses")
     result = runner.invoke(app, ["digifeeds", "load-statuses"])
     assert session_local_mock.call_count == 1
     assert load_statuse_mock.call_count == 1
@@ -87,3 +85,31 @@ def test_list_barcodes_in_input_bucket(mocker):
     assert list_barcodes_mock.call_count == 1
     assert result.exit_code == 0
     assert '["barcode1", "barcode2"]' == result.stdout
+
+
+@responses.activate
+def test_check_zephir_for_item_when_item_is_in_zephir(item_data):
+    db_url = f"{S.digifeeds_api_url}/items/some_barcode"
+    get_item = responses.get(db_url, json=item_data, status=200)
+    add_item_status = responses.put(
+        f"{db_url}/status/in_zephir", json=item_data, status=200
+    )
+    responses.get(f"{S.zephir_bib_api_url}/mdp.some_barcode", json={}, status=200)
+    result = runner.invoke(app, ["digifeeds", "check-zephir", "some_barcode"])
+    assert get_item.call_count == 1
+    assert add_item_status.call_count == 1
+    assert result.exit_code == 0
+    assert "some_barcode is in Zephir" in result.stdout
+
+
+@responses.activate
+def test_check_zephir_for_item_when_item_is_not_in_zephir(item_data):
+    db_url = f"{S.digifeeds_api_url}/items/some_barcode"
+    get_item = responses.get(db_url, json=item_data, status=200)
+    add_item_status = responses.put(f"{db_url}/status/in_zephir")
+    responses.get(f"{S.zephir_bib_api_url}/mdp.some_barcode", json={}, status=404)
+    result = runner.invoke(app, ["digifeeds", "check-zephir", "some_barcode"])
+    assert get_item.call_count == 1
+    assert add_item_status.call_count == 0
+    assert result.exit_code == 0
+    assert "some_barcode is NOT in Zephir" in result.stdout

tests/digifeeds/test_check_zephir.py

@@ -0,0 +1,72 @@
+import pytest
+import responses
+import json
+from aim.services import S
+from aim.digifeeds.check_zephir import check_zephir
+
+
+@pytest.fixture
+def item_data():
+    with open("tests/fixtures/digifeeds/item.json") as f:
+        output = json.load(f)
+    return output
+
+
+@pytest.fixture
+def barcode():
+    return "some_barcode"
+
+
+@responses.activate
+def test_barcode_is_in_zephir(mocker, item_data, barcode):
+    get_item_mock = mocker.patch(
+        "aim.digifeeds.check_zephir.DBClient.get_item", return_value=item_data
+    )
+    add_status_mock = mocker.patch(
+        "aim.digifeeds.check_zephir.DBClient.add_item_status", return_value=item_data
+    )
+
+    responses.get(f"{S.zephir_bib_api_url}/mdp.{barcode}", json={}, status=200)
+    result = check_zephir(barcode)
+    get_item_mock.assert_called_once()
+    add_status_mock.assert_called_once()
+    assert result.barcode == barcode
+
+
+def test_barcode_already_has_in_zephir_status(mocker, item_data, barcode):
+    item_data["statuses"][0]["name"] = "in_zephir"
+    get_item_mock = mocker.patch(
+        "aim.digifeeds.check_zephir.DBClient.get_item", return_value=item_data
+    )
+
+    add_status_mock = mocker.patch(
+        "aim.digifeeds.check_zephir.DBClient.add_item_status", return_value=item_data
+    )
+
+    result = check_zephir(barcode)
+    get_item_mock.assert_called_once()
+    add_status_mock.assert_not_called()
+    assert result.barcode == barcode
+
+
+@responses.activate
+def test_barcode_not_in_zephir(mocker, item_data, barcode):
+    get_item_mock = mocker.patch(
+        "aim.digifeeds.check_zephir.DBClient.get_item", return_value=item_data
+    )
+    add_status_mock = mocker.patch(
+        "aim.digifeeds.check_zephir.DBClient.add_item_status", return_value=item_data
+    )
+
+    responses.get(f"{S.zephir_bib_api_url}/mdp.{barcode}", json={}, status=404)
+    result = check_zephir(barcode)
+    get_item_mock.assert_called_once()
+    add_status_mock.assert_not_called()
+    assert result is None
+
+
+def test_barcdoe_is_not_in_db(mocker, barcode):
+    mocker.patch("aim.digifeeds.check_zephir.DBClient.get_item", return_value=None)
+    with pytest.raises(Exception) as exc_info:
+        check_zephir(barcode)
+    assert exc_info.type is Exception