Fall back to search by title

src/onepasswordconnectsdk/async_client.py

@@ -19,7 +19,7 @@ class AsyncClient:
 
     def __init__(self, url: str, token: str, config: Optional[ClientConfig] = None) -> None:
         """Initialize async client
-        
+
         Args:
             url (str): The url of the 1Password Connect API
             token (str): The 1Password Service Account token
@@ -34,11 +34,11 @@ def __init__(self, url: str, token: str, config: Optional[ClientConfig] = None)
     def create_session(self, url: str, token: str) -> httpx.AsyncClient:
         headers = self.build_headers(token)
         timeout = get_timeout()
-        
+
         if self.config:
             client_args = self.config.get_client_args(url, headers, timeout)
             return httpx.AsyncClient(**client_args)
-            
+
         return httpx.AsyncClient(base_url=url, headers=headers, timeout=timeout)
 
     def build_headers(self, token: str) -> Dict[str, str]:
@@ -115,10 +115,14 @@ async def get_item(self, item: str, vault: str) -> Item:
             vault = await self.get_vault_by_title(vault)
             vault_id = vault.id
 
-        if is_valid_uuid(item):
-            return await self.get_item_by_id(item, vault_id)
-        else:
+        if not is_valid_uuid(item):
             return await self.get_item_by_title(item, vault_id)
+        try:
+            return await self.get_item_by_id(item, vault_id)
+        except FailedToRetrieveItemException as exc:
+            if exc.status_code == 404:
+                return await self.get_item_by_title(item, vault_id)
+            raise
 
     async def get_item_by_id(self, item_id: str, vault_id: str) -> Item:
         """Get a specific item by uuid
@@ -141,7 +145,8 @@ async def get_item_by_id(self, item_id: str, vault_id: str) -> Item:
         except HTTPError:
             raise FailedToRetrieveItemException(
                 f"Unable to retrieve item. Received {response.status_code}\
-                     for {url} with message: {response.json().get('message')}"
+                     for {url} with message: {response.json().get('message')}",
+                status_code=response.status_code,
             )
         return self.serializer.deserialize(response.content, "Item")
 

src/onepasswordconnectsdk/client.py

@@ -27,7 +27,7 @@ class Client:
 
     def __init__(self, url: str, token: str, config: Optional[ClientConfig] = None) -> None:
         """Initialize client
-        
+
         Args:
             url (str): The url of the 1Password Connect API
             token (str): The 1Password Service Account token
@@ -42,11 +42,11 @@ def __init__(self, url: str, token: str, config: Optional[ClientConfig] = None)
     def create_session(self, url: str, token: str) -> httpx.Client:
         headers = self.build_headers(token)
         timeout = get_timeout()
-        
+
         if self.config:
             client_args = self.config.get_client_args(url, headers, timeout)
             return httpx.Client(**client_args)
-            
+
         return httpx.Client(base_url=url, headers=headers, timeout=timeout)
 
     def build_headers(self, token: str) -> Dict[str, str]:
@@ -122,10 +122,14 @@ def get_item(self, item: str, vault: str) -> Item:
         if not is_valid_uuid(vault):
             vault_id = self.get_vault_by_title(vault).id
 
-        if is_valid_uuid(item):
-            return self.get_item_by_id(item, vault_id)
-        else:
+        if not is_valid_uuid(item):
             return self.get_item_by_title(item, vault_id)
+        try:
+            return self.get_item_by_id(item, vault_id)
+        except FailedToRetrieveItemException as exc:
+            if exc.status_code == 404:
+                return self.get_item_by_title(item, vault_id)
+            raise
 
     def get_item_by_id(self, item_id: str, vault_id: str) -> Item:
         """Get a specific item by uuid
@@ -148,7 +152,8 @@ def get_item_by_id(self, item_id: str, vault_id: str) -> Item:
         except HTTPError:
             raise FailedToRetrieveItemException(
                 f"Unable to retrieve item. Received {response.status_code}\
-                     for {url} with message: {response.json().get('message')}"
+                     for {url} with message: {response.json().get('message')}",
+                status_code=response.status_code,
             )
         return self.serializer.deserialize(response.content, "Item")
 
@@ -398,13 +403,13 @@ def sanitize_for_serialization(self, obj):
 
 def new_client(url: str, token: str, is_async: bool = False, config: Optional[ClientConfig] = None) -> Union[AsyncClient, Client]:
     """Builds a new client for interacting with 1Password Connect
-    
+
     Args:
         url (str): The url of the 1Password Connect API
         token (str): The 1Password Service Account token
         is_async (bool): Initialize async or sync client
         config (Optional[ClientConfig]): Optional configuration for httpx client
-        
+
     Returns:
         Union[AsyncClient, Client]: The 1Password Connect client
     """

src/onepasswordconnectsdk/errors.py

@@ -11,7 +11,9 @@ class EnvironmentHostNotSetException(OnePasswordConnectSDKError, TypeError):
 
 
 class FailedToRetrieveItemException(OnePasswordConnectSDKError):
-    pass
+    def __init__(self, message, *, status_code=None):
+        super().__init__(message)
+        self.status_code = status_code
 
 
 class FailedToRetrieveVaultException(OnePasswordConnectSDKError):

src/onepasswordconnectsdk/utils.py

@@ -7,7 +7,7 @@
 
 
 def is_valid_uuid(uuid):
-    if len(uuid) is not UUIDLength:
+    if len(uuid) != UUIDLength:
         return False
     for c in uuid:
         valid = (c >= 'a' and c <= 'z') or (c >= '0' and c <= '9')

src/tests/test_client_items.py

@@ -11,6 +11,8 @@
 VAULT_TITLE = "VaultA"
 ITEM_ID = "wepiqdxdzncjtnvmv5fegud4qy"
 ITEM_TITLE = "Test Login"
+# 26 lowercase alphanumeric chars: treated as item id by is_valid_uuid but may be a title (#80)
+ITEM_TITLE_26_CHARS = "abcdefghijklmnop1234567890"
 HOST = "https://mock_host"
 TOKEN = "jwt_token"
 SS_CLIENT = client.new_client(HOST, TOKEN)
@@ -193,6 +195,58 @@ async def test_get_item_by_item_title_vault_title_async(respx_mock):
     assert item_mock.called
 
 
+def test_get_item_26_char_title_falls_back_from_id_to_title(respx_mock):
+    """Item titles matching the SDK item-id shape should resolve via title after 404 on id."""
+    expected_item = get_item()
+    expected_path_by_id = f"/v1/vaults/{VAULT_ID}/items/{ITEM_TITLE_26_CHARS}"
+    expected_path_by_title = (
+        f'/v1/vaults/{VAULT_ID}/items?filter=title eq "{ITEM_TITLE_26_CHARS}"'
+    )
+    expected_path_item = f"/v1/vaults/{VAULT_ID}/items/{ITEM_ID}"
+
+    by_id_mock = respx_mock.get(expected_path_by_id).mock(
+        return_value=Response(404, json={"message": "not found"})
+    )
+    items_by_title_mock = respx_mock.get(expected_path_by_title).mock(
+        return_value=Response(200, json=get_items_with_title(ITEM_TITLE_26_CHARS))
+    )
+    item_mock = respx_mock.get(expected_path_item).mock(
+        return_value=Response(200, json=expected_item)
+    )
+
+    item = SS_CLIENT.get_item(ITEM_TITLE_26_CHARS, VAULT_ID)
+    compare_items(expected_item, item)
+    assert by_id_mock.called
+    assert items_by_title_mock.called
+    assert item_mock.called
+
+
+@pytest.mark.asyncio
+async def test_get_item_26_char_title_falls_back_from_id_to_title_async(respx_mock):
+    expected_item = get_item()
+    expected_path_by_id = f"/v1/vaults/{VAULT_ID}/items/{ITEM_TITLE_26_CHARS}"
+    expected_path_by_title = (
+        f'/v1/vaults/{VAULT_ID}/items?filter=title eq "{ITEM_TITLE_26_CHARS}"'
+    )
+    expected_path_item = f"/v1/vaults/{VAULT_ID}/items/{ITEM_ID}"
+
+    by_id_mock = respx_mock.get(expected_path_by_id).mock(
+        return_value=Response(404, json={"message": "not found"})
+    )
+    items_by_title_mock = respx_mock.get(expected_path_by_title).mock(
+        return_value=Response(200, json=get_items_with_title(ITEM_TITLE_26_CHARS))
+    )
+    item_mock = respx_mock.get(expected_path_item).mock(
+        return_value=Response(200, json=expected_item)
+    )
+
+    item = await SS_CLIENT_ASYNC.get_item(ITEM_TITLE_26_CHARS, VAULT_ID)
+    compare_items(expected_item, item)
+    assert by_id_mock.called
+    assert items_by_title_mock.called
+    assert item_mock.called
+
+
 def test_get_items(respx_mock):
     expected_items = get_items()
     expected_path = f"/v1/vaults/{VAULT_ID}/items"
@@ -346,6 +400,12 @@ def compare_sections(expected_section, returned_section):
     assert expected_section.get("label") == returned_section.label
 
 
+def get_items_with_title(title: str):
+    row = dict(get_items()[0])
+    row["title"] = title
+    return [row]
+
+
 def get_items():
     return [{
         "id": "wepiqdxdzncjtnvmv5fegud4qy",