code for building cell type and region hierarchies

Author: Clare72

src/test/test_hierarchy.py

@@ -0,0 +1,133 @@
+"""Tests for get_hierarchy function.
+
+Tests the hierarchy tree builder for both part_of (brain region structure)
+and subclass_of (cell type hierarchies), in both ancestor and descendant
+directions.
+"""
+
+import pytest
+
+from vfbquery.vfb_queries import get_hierarchy
+
+
+# Known test terms
+MUSHROOM_BODY = "FBbt_00005801"
+KENYON_CELL = "FBbt_00003686"
+
+
+class TestHierarchyValidation:
+    def test_invalid_relationship_raises(self):
+        with pytest.raises(ValueError, match="relationship"):
+            get_hierarchy(KENYON_CELL, relationship="invalid")
+
+    def test_invalid_direction_raises(self):
+        with pytest.raises(ValueError, match="direction"):
+            get_hierarchy(KENYON_CELL, direction="invalid")
+
+
+class TestSubclassOfDescendants:
+    @pytest.mark.integration
+    def test_returns_descendants(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        assert result['id'] == KENYON_CELL
+        assert result['label'] == 'Kenyon cell'
+        assert result['relationship'] == 'subclass_of'
+        assert 'descendants' in result
+        assert len(result['descendants']) > 0
+
+    @pytest.mark.integration
+    def test_descendants_have_id_and_label(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        for child in result['descendants']:
+            assert 'id' in child
+            assert 'label' in child
+            assert child['id'].startswith('FBbt_')
+            assert child['label'] != child['id']  # label should be resolved
+
+    @pytest.mark.integration
+    def test_depth_1_has_no_grandchildren(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        for child in result['descendants']:
+            assert 'descendants' not in child
+
+    @pytest.mark.integration
+    def test_depth_2_has_nested_children(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=2)
+        has_grandchildren = any('descendants' in child for child in result['descendants'])
+        assert has_grandchildren, "At least one direct subclass should have its own subclasses"
+
+
+class TestSubclassOfAncestors:
+    @pytest.mark.integration
+    def test_returns_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        assert 'ancestors' in result
+        assert len(result['ancestors']) > 0
+
+    @pytest.mark.integration
+    def test_ancestors_have_id_and_label(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        for anc in result['ancestors']:
+            assert 'id' in anc
+            assert 'label' in anc
+
+    @pytest.mark.integration
+    def test_kenyon_cell_ancestor_is_mb_intrinsic_neuron(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        ancestor_ids = [a['id'] for a in result['ancestors']]
+        assert 'FBbt_00007484' in ancestor_ids  # mushroom body intrinsic neuron
+
+    @pytest.mark.integration
+    def test_depth_2_has_nested_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=2)
+        has_grandparent = any('ancestors' in anc for anc in result['ancestors'])
+        assert has_grandparent
+
+
+class TestPartOfDescendants:
+    @pytest.mark.integration
+    def test_returns_parts(self):
+        result = get_hierarchy(MUSHROOM_BODY, 'part_of', 'descendants', max_depth=1)
+        assert result['id'] == MUSHROOM_BODY
+        assert result['label'] == 'mushroom body'
+        assert 'descendants' in result
+        assert len(result['descendants']) > 0
+
+    @pytest.mark.integration
+    def test_parts_have_id_and_label(self):
+        result = get_hierarchy(MUSHROOM_BODY, 'part_of', 'descendants', max_depth=1)
+        for part in result['descendants']:
+            assert 'id' in part
+            assert 'label' in part
+            assert part['id'].startswith('FBbt_')
+
+
+class TestPartOfAncestors:
+    @pytest.mark.integration
+    def test_mushroom_body_part_of_protocerebrum(self):
+        result = get_hierarchy(MUSHROOM_BODY, 'part_of', 'ancestors', max_depth=1)
+        assert 'ancestors' in result
+        ancestor_ids = [a['id'] for a in result['ancestors']]
+        assert 'FBbt_00003627' in ancestor_ids  # protocerebrum
+
+
+class TestBothDirections:
+    @pytest.mark.integration
+    def test_both_returns_ancestors_and_descendants(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'both', max_depth=1)
+        assert 'ancestors' in result
+        assert 'descendants' in result
+        assert len(result['ancestors']) > 0
+        assert len(result['descendants']) > 0
+
+    @pytest.mark.integration
+    def test_descendants_only_has_no_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        assert 'descendants' in result
+        assert 'ancestors' not in result
+
+    @pytest.mark.integration
+    def test_ancestors_only_has_no_descendants(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        assert 'ancestors' in result
+        assert 'descendants' not in result

src/vfbquery/ha_api.py

@@ -897,6 +897,39 @@ def post_fn(result):
     )
 
 
+def _run_get_hierarchy(short_form, relationship, direction, max_depth):
+    """Worker: run get_hierarchy in a subprocess."""
+    from . import vfb_queries as _vfb
+    return _convert_numpy_types(
+        _vfb.get_hierarchy(short_form, relationship=relationship,
+                           direction=direction, max_depth=max_depth)
+    )
+
+
+async def handle_get_hierarchy(request):
+    """GET /get_hierarchy?id=FBbt_00005801&relationship=part_of&direction=both&max_depth=1"""
+    short_form = request.query.get("id")
+    if not short_form:
+        return web.json_response({"error": "id parameter is required"}, status=400)
+    relationship = request.query.get("relationship", "part_of")
+    if relationship not in ("part_of", "subclass_of"):
+        return web.json_response(
+            {"error": "relationship must be 'part_of' or 'subclass_of'"}, status=400
+        )
+    direction = request.query.get("direction", "both")
+    if direction not in ("descendants", "ancestors", "both"):
+        return web.json_response(
+            {"error": "direction must be 'descendants', 'ancestors', or 'both'"}, status=400
+        )
+    max_depth = int(request.query.get("max_depth", "1"))
+
+    key = f"get_hierarchy:{short_form}:{relationship}:{direction}:{max_depth}"
+    return await _dispatch_to_pool(
+        request, key, _run_get_hierarchy,
+        short_form, relationship, direction, max_depth,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Application factory
 # ---------------------------------------------------------------------------
@@ -937,6 +970,7 @@ def create_app(max_workers=None, max_concurrent=None, max_queue_depth=None,
     app.router.add_get("/find_combo_publications", handle_find_combo_publications)
     app.router.add_get("/list_connectome_datasets", handle_list_connectome_datasets)
     app.router.add_get("/query_connectivity", handle_query_connectivity)
+    app.router.add_get("/get_hierarchy", handle_get_hierarchy)
 
     # Store config for /status and handlers
     app["max_workers"] = max_workers

src/vfbquery/owlery_client.py

@@ -105,7 +105,7 @@ def convert_short_form_to_iri(match):
             # Based on VFBConnect's query() method
             params = {
                 'object': iri_query,
-                'direct': 'false',  # Always use indirect (transitive) queries
+                'direct': 'true' if direct else 'false',
                 'includeDeprecated': 'false',  # Exclude deprecated terms
                 'includeEquivalent': 'true'  # Include equivalent classes
             }