REGI-500: Lookups

cwms/locations/lookups.py

@@ -0,0 +1,162 @@
+#  Copyright (c) 2026
+#  United States Army Corps of Engineers - Hydrologic Engineering Center (USACE/HEC)
+#  All Rights Reserved.  USACE PROPRIETARY/CONFIDENTIAL.
+#  Source may not be released without written approval from HEC
+
+import cwms.api as api
+from cwms.cwms_types import JSON, Data
+
+ENDPOINT = "lookup-types"
+
+
+def get_all_lookups(category: str, prefix: str, office_id: str) -> Data:
+    """
+    Retrieves all lookups for a given category, prefix, and office.
+
+    Parameters
+    ----------
+    category : str
+        Filters lookup types to the specified category
+    prefix : str
+        Filters lookup types to the specified prefix
+    office_id : str
+        Filters lookup types to the specified office ID
+
+    Returns
+    -------
+    Data
+        The JSON response from CWMS Data API wrapped in a Data object.
+
+    Raises
+    ------
+    ValueError
+        If any required argument is missing.
+    ClientError
+        If a 400 range error code response is returned from the server.
+    NoDataFoundError
+        If a 404 range error code response is returned from the server.
+    ServerError
+        If a 500 range error code response is returned from the server.
+    """
+    if not all([category, prefix, office_id]):
+        raise ValueError("Category, Prefix, and Office ID must be specified")
+
+    params = {"category": category, "prefix": prefix, "office": office_id}
+    response = api.get(ENDPOINT, params, api_version=1)
+    return Data(response)
+
+
+def create_lookup(data: JSON, category: str, prefix: str) -> None:
+    """
+    Creates a new lookup entry.
+
+    Parameters
+    ----------
+    data: JSON
+        A dictionary representing the JSON data to be stored. This should match the
+        LookupType structure as defined by the API.
+    category : str
+        Specifies the category of the lookup.
+    prefix : str
+        Specifies the prefix of the lookup.
+
+    Returns
+    -------
+    None
+
+    Raises
+    ------
+    ValueError
+        If any required argument is missing.
+    ClientError
+        If a 400 range error code response is returned from the server.
+    NoDataFoundError
+        If a 404 range error code response is returned from the server.
+    ServerError
+        If a 500 range error code response is returned from the server.
+    """
+    if not all([category, prefix]):
+        raise ValueError("Category and Prefix must be specified")
+    if not data:
+        raise ValueError("Data must be specified")
+    params = {"category": category, "prefix": prefix}
+    api.post(ENDPOINT, data, params, api_version=1)
+
+
+def update_lookup(data: JSON, name: str, category: str, prefix: str) -> None:
+    """
+    Updates a specified lookup entry.
+
+    Parameters
+    ----------
+    data : JSON
+        A dictionary representing the JSON data to be stored.
+        If the `data` value is None, a `ValueError` will be raised.
+    name : str
+        Specifies the location type to update
+    category : str
+        Specifies the category of the lookup.
+    prefix : str
+        Specifies the prefix of the lookup.
+
+    Returns
+    -------
+    None
+
+    Raises
+    ------
+    ValueError
+        If any required argument is missing.
+    ClientError
+        If a 400 range error code response is returned from the server.
+    NoDataFoundError
+        If a 404 range error code response is returned from the server.
+    ServerError
+        If a 500 range error code response is returned from the server.
+    """
+    if not all([name, category, prefix]):
+        raise ValueError("Name, Category, and Prefix must be specified")
+    if not data:
+        raise ValueError("Data must be specified")
+
+    endpoint = f"{ENDPOINT}/{name}"
+    params = {"category": category, "prefix": prefix}
+    api.patch(endpoint, data, params, api_version=1)
+
+
+def delete_lookup(name: str, category: str, prefix: str, office_id: str) -> None:
+    """
+    Deletes a specified lookup entry.
+
+    Parameters
+    ----------
+    name : str
+        Specifies the location type to delete.
+    category : str
+        Specifies the category id of the lookup type to be deleted.
+    prefix : str
+        Specifies the prefix of the lookup type to be deleted.
+    office_id : str
+        Specifies the owning office of the lookup type to be deleted.
+
+    Returns
+    -------
+    None
+
+    Raises
+    ------
+    ValueError
+        If any required argument is missing.
+    ClientError
+        If a 400 range error code response is returned from the server.
+    NoDataFoundError
+        If a 404 range error code response is returned from the server.
+    ServerError
+        If a 500 range error code response is returned from the server.
+    """
+    if not all([name, category, prefix, office_id]):
+        raise ValueError("Name, Category, Prefix, and Office ID must be specified")
+
+    endpoint = f"{ENDPOINT}/{name}"
+    params = {"category": category, "prefix": prefix, "office": office_id}
+    api.delete(endpoint, params, api_version=1)

tests/cda/locations/lookup_test.py

@@ -0,0 +1,127 @@
+import pytest
+
+import cwms.api as api
+import cwms.locations.lookups as lookups
+from cwms.cwms_types import Data
+
+OFFICE_ID = "SPK"
+CATEGORY = "USACE"
+PREFIX = "L"
+DISPLAY_VALUE = "Test Lookup"
+TOOLTIP = "Test Tooltip"
+
+LOOKUP_DATA = {
+    "office-id": OFFICE_ID,
+    "display-value": DISPLAY_VALUE,
+    "tooltip": TOOLTIP,
+    "active": True,
+}
+
+
+def _cleanup():
+    try:
+        lookups.delete_lookup(DISPLAY_VALUE, CATEGORY, PREFIX, OFFICE_ID)
+    except Exception:
+        pass
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_data():
+    _cleanup()
+    lookups.create_lookup(LOOKUP_DATA, CATEGORY, PREFIX)
+
+
+@pytest.fixture(autouse=True)
+def init_session():
+    print("Initializing CWMS API session for lookup type tests...")
+
+
+def test_get_all_lookups():
+    new_display_value = DISPLAY_VALUE + " Refreshed"
+    data = {
+        "office-id": OFFICE_ID,
+        "display-value": new_display_value,
+        "tooltip": TOOLTIP,
+        "active": True,
+    }
+    lookups.create_lookup(data, CATEGORY, PREFIX)
+
+    result = lookups.get_all_lookups(CATEGORY, PREFIX, OFFICE_ID)
+    found = False
+    found2 = False
+    for item in result.json:
+        if item.get("display-value") == new_display_value:
+            found = True
+        if item.get("display-value") == DISPLAY_VALUE:
+            found2 = True
+    assert (
+        found
+    ), f"Lookup with display-value {new_display_value} not found after creation"
+    assert found2, f"Lookup with display-value {DISPLAY_VALUE} not found after creation"
+
+
+def test_create_lookup():
+    display_value = DISPLAY_VALUE + " Created"
+    data = {
+        "office-id": OFFICE_ID,
+        "display-value": display_value,
+        "tooltip": TOOLTIP,
+        "active": True,
+    }
+    lookups.create_lookup(data, CATEGORY, PREFIX)
+
+    result = lookups.get_all_lookups(CATEGORY, PREFIX, OFFICE_ID)
+    found = False
+    for item in result.json:
+        if item.get("display-value") == display_value:
+            found = True
+            break
+    assert found, f"Lookup with display-value {display_value} not found after creation"
+
+
+def test_update_lookup():
+    new_display_value = DISPLAY_VALUE + " Updated"
+    data = {
+        "office-id": OFFICE_ID,
+        "display-value": new_display_value,
+        "tooltip": "Updated Tooltip",
+        "active": True,
+    }
+    # The name parameter in update_lookup corresponds to the display-value of the item to update
+    lookups.update_lookup(data, DISPLAY_VALUE, CATEGORY, PREFIX)
+
+    result = lookups.get_all_lookups(CATEGORY, PREFIX, OFFICE_ID)
+    found = False
+    for item in result.json:
+        if item.get("display-value") == new_display_value:
+            found = True
+            break
+    assert found, f"Lookup with updated display-value {new_display_value} not found"
+
+    try:
+        lookups.delete_lookup(new_display_value, CATEGORY, PREFIX, OFFICE_ID)
+    except Exception:
+        pass
+
+
+def test_delete_lookup():
+    office = "LRL"
+    data = {
+        "office-id": office,
+        "display-value": DISPLAY_VALUE,
+        "tooltip": TOOLTIP,
+        "active": True,
+    }
+    lookups.create_lookup(data, CATEGORY, PREFIX)
+
+    lookups.delete_lookup(DISPLAY_VALUE, CATEGORY, PREFIX, office)
+
+    result = lookups.get_all_lookups(CATEGORY, PREFIX, office)
+    found = False
+    for item in result.json:
+        if item.get("display-value") == DISPLAY_VALUE:
+            found = True
+            break
+    assert (
+        not found
+    ), f"Lookup with display-value {DISPLAY_VALUE} still found after deletion"