feat(search): add iscc_id lookup with self-exclusion

Implement iscc_id-based similarity search across all index backends
(memory, lmdb, usearch). When iscc_id is provided in query, the system
looks up the asset and uses its iscc_code/units/simprints for similarity
search while excluding the query asset from results.

- Add iscc_id query handling with precedence over other fields
- Implement self-exclusion to prevent query asset in results
- Return 404 when iscc_id not found (FileNotFoundError)
- Add comprehensive tests for iscc_id search behavior
- Update existing tests to avoid iscc_id conflicts

Author: titusz

iscc_search/indexes/lmdb/index.py

@@ -204,6 +204,17 @@ def search_assets(self, query, limit=100):
         :return: IsccSearchResult with matches sorted by score (descending, normalized 0.0-1.0)
         :raises ValueError: If query has neither iscc_code nor units
         """
+        # Handle iscc_id lookup if provided (takes precedence over other fields)
+        query_iscc_id = None  # Track original query iscc_id for self-exclusion
+        if query.iscc_id:
+            query_iscc_id = query.iscc_id
+            # Look up asset by iscc_id (raises FileNotFoundError if not found -> HTTP 404)
+            asset = self.get_asset(query.iscc_id)
+            # Create new query with extracted iscc_code, units and simprints
+            from iscc_search.schema import IsccQuery
+
+            query = IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=asset.simprints)
+
         # Normalize query to ensure it has units (derive from iscc_code if needed)
         query = common.normalize_query(query)
 
@@ -273,6 +284,10 @@ def search_assets(self, query, limit=100):
             # Sort by score descending
             match_list.sort(key=lambda x: x.score, reverse=True)
 
+            # Exclude query asset from results (self-exclusion for iscc_id queries)
+            if query_iscc_id:
+                match_list = [match for match in match_list if match.iscc_id != query_iscc_id]
+
             return IsccSearchResult(query=query, global_matches=match_list[:limit], chunk_matches=[])
 
     def get_asset_count(self):

iscc_search/indexes/memory/index.py

@@ -190,6 +190,17 @@ def search_assets(self, index_name, query, limit=100):
         if index_name not in self._indexes:
             raise FileNotFoundError(f"Index '{index_name}' not found")
 
+        # Handle iscc_id lookup if provided (takes precedence over other fields)
+        query_iscc_id = None  # Track original query iscc_id for self-exclusion
+        if query.iscc_id:
+            query_iscc_id = query.iscc_id
+            # Look up asset by iscc_id (raises FileNotFoundError if not found -> HTTP 404)
+            asset = self.get_asset(index_name, query.iscc_id)
+            # Create new query with extracted iscc_code, units and simprints
+            from iscc_search.schema import IsccQuery
+
+            query = IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=asset.simprints)
+
         # Normalize query to ensure it has units (derive from iscc_code if needed)
         # This ensures consistent behavior across backends
         query = common.normalize_query(query)
@@ -211,6 +222,10 @@ def search_assets(self, index_name, query, limit=100):
                         )
                     )
 
+        # Exclude query asset from results (self-exclusion for iscc_id queries)
+        if query_iscc_id:
+            match_list = [match for match in match_list if match.iscc_id != query_iscc_id]
+
         return IsccSearchResult(
             query=query,
             global_matches=match_list[:limit],

iscc_search/indexes/usearch/index.py

@@ -343,6 +343,17 @@ def search_assets(self, query, limit=100):
         :param limit: Maximum number of results
         :return: IsccSearchResult with query and list of matches (scores normalized 0.0-1.0)
         """
+        # Handle iscc_id lookup if provided (takes precedence over other fields)
+        query_iscc_id = None  # Track original query iscc_id for self-exclusion
+        if query.iscc_id:
+            query_iscc_id = query.iscc_id
+            # Look up asset by iscc_id (raises FileNotFoundError if not found -> HTTP 404)
+            asset = self.get_asset(query.iscc_id)
+            # Create new query with extracted iscc_code, units and simprints
+            from iscc_search.schema import IsccQuery
+
+            query = IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=asset.simprints)
+
         # Normalize query
         query = common.normalize_query(query)
 
@@ -399,6 +410,11 @@ def search_assets(self, query, limit=100):
 
                 scored_results.append((key, total_score, unit_scores))
 
+            # Exclude query asset from results (self-exclusion for iscc_id queries)
+            if query_iscc_id:
+                query_key = int(IsccID(query_iscc_id))
+                scored_results = [result for result in scored_results if result[0] != query_key]
+
             # Sort by total score descending
             scored_results.sort(key=lambda x: x[1], reverse=True)
 
@@ -441,6 +457,10 @@ def search_assets(self, query, limit=100):
                     iscc_id = str(IsccID.from_int(key, self._realm_id))
                     matches.append(IsccGlobalMatch(iscc_id=iscc_id, score=total_score, types=unit_scores))
 
+        # Exclude query asset from chunk matches (self-exclusion for iscc_id queries)
+        if query_iscc_id:
+            chunk_matches = [match for match in chunk_matches if match.iscc_id != query_iscc_id]
+
         return IsccSearchResult(query=query, global_matches=matches, chunk_matches=chunk_matches)
 
     def flush(self):

tests/test_indexes_lmdb_index.py

@@ -181,14 +181,6 @@ def test_search_assets_basic(lmdb_index, sample_assets):
     assert result.global_matches[0].metadata.model_dump(exclude_none=True) == asset.metadata
 
 
-def test_search_assets_no_units(lmdb_index, sample_iscc_ids):
-    """Test search without units or iscc_code raises error."""
-    query = IsccQuery(iscc_id=sample_iscc_ids[0])
-
-    with pytest.raises(ValueError, match="must have 'iscc_code', 'units', or 'simprints'"):
-        lmdb_index.search_assets(query)
-
-
 def test_search_assets_empty_index(lmdb_index, sample_content_units):
     """Test search on empty index returns no matches."""
     query = IsccQuery(units=[sample_content_units[0], sample_content_units[1]])

tests/test_indexes_memory_index.py

@@ -300,20 +300,22 @@ def test_search_assets_by_iscc_code(sample_iscc_ids, sample_iscc_codes):
 
 
 def test_search_assets_by_iscc_id(sample_iscc_ids, sample_iscc_codes):
-    """Test searching assets by iscc_id with iscc_code."""
+    """Test searching assets by iscc_id uses asset's code for similarity search and excludes query asset."""
     index = MemoryIndex()
     index.create_index(IsccIndex(name="testindex"))
 
-    # Add assets with iscc_code (required for search)
-    asset = IsccEntry(iscc_id=sample_iscc_ids[0], iscc_code=sample_iscc_codes[0])
-    index.add_assets("testindex", [asset])
+    # Add two assets with same iscc_code (similar assets)
+    asset1 = IsccEntry(iscc_id=sample_iscc_ids[0], iscc_code=sample_iscc_codes[0])
+    asset2 = IsccEntry(iscc_id=sample_iscc_ids[1], iscc_code=sample_iscc_codes[0])
+    index.add_assets("testindex", [asset1, asset2])
 
-    # Search by iscc_id and iscc_code
-    query = IsccQuery(iscc_id=sample_iscc_ids[0], iscc_code=sample_iscc_codes[0])
+    # Search by iscc_id of first asset
+    query = IsccQuery(iscc_id=sample_iscc_ids[0])
     result = index.search_assets("testindex", query)
 
+    # Should find the similar asset but exclude the query asset itself (self-exclusion)
     assert len(result.global_matches) == 1
-    assert result.global_matches[0].iscc_id == sample_iscc_ids[0]
+    assert result.global_matches[0].iscc_id == sample_iscc_ids[1]
 
 
 def test_search_assets_no_matches(sample_iscc_ids, sample_iscc_codes):
@@ -447,20 +449,20 @@ def test_metadata_field(sample_iscc_ids, sample_iscc_codes):
 
 
 def test_search_assets_no_matching_iscc_id(sample_iscc_ids, sample_iscc_codes):
-    """Test searching by different iscc_code when no match exists."""
+    """Test searching by non-existent iscc_id raises FileNotFoundError."""
     index = MemoryIndex()
     index.create_index(IsccIndex(name="testindex"))
 
-    # Add an asset with one iscc_code
+    # Add an asset with one iscc_id
     asset = IsccEntry(iscc_id=sample_iscc_ids[0], iscc_code=sample_iscc_codes[0])
     index.add_assets("testindex", [asset])
 
-    # Search with different iscc_id and iscc_code
-    query = IsccQuery(iscc_id=sample_iscc_ids[1], iscc_code=sample_iscc_codes[1])
-    result = index.search_assets("testindex", query)
+    # Search with different iscc_id (not in index)
+    query = IsccQuery(iscc_id=sample_iscc_ids[1])
 
-    # Should not match (different iscc_code)
-    assert len(result.global_matches) == 0
+    # Should raise FileNotFoundError (asset lookup fails -> HTTP 404)
+    with pytest.raises(FileNotFoundError, match=f"Asset '{sample_iscc_ids[1]}' not found"):
+        index.search_assets("testindex", query)
 
 
 def test_search_assets_no_iscc_code_in_asset(sample_iscc_ids, sample_iscc_codes):

tests/test_server_search.py

@@ -10,8 +10,8 @@ def test_search_post_success(test_client, sample_assets):
     assets_dict = [a.model_dump(mode="json", exclude_none=True) for a in sample_assets]
     test_client.post("/indexes/testindex/assets", json=assets_dict)
 
-    # Search using first asset as query
-    query_dict = sample_assets[0].model_dump(mode="json", exclude_none=True)
+    # Search using first asset as query (exclude iscc_id to avoid iscc_id search)
+    query_dict = sample_assets[0].model_dump(mode="json", exclude_none=True, exclude={"iscc_id"})
     response = test_client.post("/indexes/testindex/search", json=query_dict)
 
     assert response.status_code == 200
@@ -147,8 +147,8 @@ def test_search_result_structure(test_client, sample_assets):
     assets_dict = [a.model_dump(mode="json", exclude_none=True) for a in sample_assets]
     test_client.post("/indexes/testindex/assets", json=assets_dict)
 
-    # Search
-    query_dict = sample_assets[0].model_dump(mode="json", exclude_none=True)
+    # Search (exclude iscc_id to avoid iscc_id search)
+    query_dict = sample_assets[0].model_dump(mode="json", exclude_none=True, exclude={"iscc_id"})
     response = test_client.post("/indexes/testindex/search", json=query_dict)
 
     assert response.status_code == 200
@@ -174,8 +174,8 @@ def test_search_empty_index(test_client, sample_assets):
     # Create empty index
     test_client.post("/indexes", json={"name": "testindex"})
 
-    # Search empty index
-    query_dict = sample_assets[0].model_dump(mode="json", exclude_none=True)
+    # Search empty index (exclude iscc_id to avoid iscc_id lookup failure)
+    query_dict = sample_assets[0].model_dump(mode="json", exclude_none=True, exclude={"iscc_id"})
     response = test_client.post("/indexes/testindex/search", json=query_dict)
 
     assert response.status_code == 200
@@ -284,3 +284,85 @@ def test_response_excludes_unset_fields(test_client, request):
     assert "iscc_id" not in query
     assert "iscc_code" not in query
     assert "units" not in query
+
+
+def test_search_by_iscc_id_success(test_client, sample_assets):
+    """Test POST search with iscc_id parameter finds similar assets."""
+    # Create index and add multiple assets
+    test_client.post("/indexes", json={"name": "testindex"})
+    assets_dict = [a.model_dump(mode="json", exclude_none=True) for a in sample_assets]
+    test_client.post("/indexes/testindex/assets", json=assets_dict)
+
+    # Search using iscc_id from first asset
+    query_dict = {"iscc_id": sample_assets[0].iscc_id}
+    response = test_client.post("/indexes/testindex/search", json=query_dict)
+
+    assert response.status_code == 200
+    data = response.json()
+
+    # Verify response structure
+    assert "query" in data
+    assert "global_matches" in data
+
+    # Should find similar assets (excluding the query asset itself)
+    assert isinstance(data["global_matches"], list)
+
+
+def test_search_by_iscc_id_self_exclusion(test_client, sample_assets):
+    """Test that query asset is excluded from results when searching by iscc_id."""
+    # Create index and add multiple assets
+    test_client.post("/indexes", json={"name": "testindex"})
+    assets_dict = [a.model_dump(mode="json", exclude_none=True) for a in sample_assets]
+    test_client.post("/indexes/testindex/assets", json=assets_dict)
+
+    # Search using iscc_id from first asset
+    query_iscc_id = sample_assets[0].iscc_id
+    query_dict = {"iscc_id": query_iscc_id}
+    response = test_client.post("/indexes/testindex/search", json=query_dict)
+
+    assert response.status_code == 200
+    data = response.json()
+
+    # Verify query asset is NOT in results
+    result_iscc_ids = [match["iscc_id"] for match in data["global_matches"]]
+    assert query_iscc_id not in result_iscc_ids
+
+
+def test_search_by_iscc_id_not_found(test_client, sample_iscc_ids):
+    """Test POST search with iscc_id that doesn't exist returns 404."""
+    # Create empty index
+    test_client.post("/indexes", json={"name": "testindex"})
+
+    # Search for non-existent iscc_id
+    query_dict = {"iscc_id": sample_iscc_ids[0]}
+    response = test_client.post("/indexes/testindex/search", json=query_dict)
+
+    # Should return 404 as documented
+    assert response.status_code == 404
+    data = response.json()
+    assert "detail" in data
+    assert sample_iscc_ids[0] in data["detail"]
+
+
+def test_search_by_iscc_id_precedence(test_client, sample_assets, sample_iscc_codes):
+    """Test that iscc_id takes precedence over other query fields."""
+    # Create index and add assets
+    test_client.post("/indexes", json={"name": "testindex"})
+    assets_dict = [a.model_dump(mode="json", exclude_none=True) for a in sample_assets]
+    test_client.post("/indexes/testindex/assets", json=assets_dict)
+
+    # Search with iscc_id AND other fields (iscc_id should take precedence)
+    query_dict = {
+        "iscc_id": sample_assets[0].iscc_id,
+        "iscc_code": sample_iscc_codes[5],  # Different code - should be ignored
+        "units": ["ISCC:AAAUHBUDQUT3LPWR"],  # Different units - should be ignored
+    }
+    response = test_client.post("/indexes/testindex/search", json=query_dict)
+
+    assert response.status_code == 200
+    data = response.json()
+
+    # Verify query asset is excluded (self-exclusion for iscc_id)
+    query_iscc_id = sample_assets[0].iscc_id
+    result_iscc_ids = [match["iscc_id"] for match in data["global_matches"]]
+    assert query_iscc_id not in result_iscc_ids