[ENH] Add endpoint returning queryable attributes + refine existing endpoint for attribute instances

app/api/crud.py

@@ -192,9 +192,52 @@ async def get_terms(data_element_URI: str):
 
     results_dict = {
         data_element_URI: [
-            result["termURL"]["value"]
+            util.replace_namespace_uri(result["termURL"]["value"])
             for result in results["results"]["bindings"]
         ]
     }
 
     return results_dict
+
+
+async def get_controlled_term_attributes():
+    """
+    Makes a POST query to Stardog API for all Neurobagel classes representing controlled term attributes.
+
+    Returns
+    -------
+    dict
+        Dictionary with value corresponding to all available controlled term attributes.
+    """
+    attributes_query = f"""
+    {util.DEFAULT_CONTEXT}
+
+    SELECT DISTINCT ?attribute
+    WHERE {{
+        ?attribute rdfs:subClassOf nb:ControlledTerm .
+    }}
+    """
+
+    response = httpx.post(
+        url=util.QUERY_URL,
+        content=attributes_query,
+        headers=util.QUERY_HEADER,
+        auth=httpx.BasicAuth(
+            os.environ.get(util.GRAPH_USERNAME.name),
+            os.environ.get(util.GRAPH_PASSWORD.name),
+        ),
+    )
+
+    if not response.is_success:
+        raise HTTPException(
+            status_code=response.status_code,
+            detail=f"{response.reason_phrase}: {response.text}",
+        )
+
+    results = response.json()
+    results_list = [
+        util.replace_namespace_uri(result["attribute"]["value"])
+        for result in results["results"]["bindings"]
+    ]
+
+    return results_list

app/api/routers/attributes.py

@@ -0,0 +1,23 @@
+from fastapi import APIRouter
+from pydantic import constr
+
+from .. import crud
+from ..models import CONTROLLED_TERM_REGEX
+
+router = APIRouter(prefix="/attributes", tags=["attributes"])
+
+
+@router.get("/{data_element_URI}")
+async def get_terms(data_element_URI: constr(regex=CONTROLLED_TERM_REGEX)):
+    """When a GET request is sent, return a dict with the only key corresponding to controlled term of a neurobagel class and value corresponding to all the available terms."""
+    response = await crud.get_terms(data_element_URI)
+
+    return response
+
+
+@router.get("/", response_model=list)
+async def get_attributes():
+    """When a GET request is sent, return a list of the harmonized controlled term attributes."""
+    response = await crud.get_controlled_term_attributes()
+
+    return response

app/api/routers/query.py

@@ -3,10 +3,9 @@
 from typing import List
 
 from fastapi import APIRouter, Depends
-from pydantic import constr
 
 from .. import crud
-from ..models import CONTROLLED_TERM_REGEX, CohortQueryResponse, QueryModel
+from ..models import CohortQueryResponse, QueryModel
 
 router = APIRouter(prefix="/query", tags=["query"])
 
@@ -26,11 +25,3 @@ async def get_query(query: QueryModel = Depends(QueryModel)):
     )
 
     return response
-
-
-@router.get("/attributes/{data_element_URI}")
-async def get_terms(data_element_URI: constr(regex=CONTROLLED_TERM_REGEX)):
-    """When a GET request is sent, return a dict with the only key corresponding to controlled term of a neurobagel class and value corresponding to all the available terms."""
-    response = await crud.get_terms(data_element_URI)
-
-    return response

app/api/utility.py

@@ -38,6 +38,7 @@
 }
 
 # SPARQL query context
+# TODO: Refactor into a function.
 DEFAULT_CONTEXT = """
 PREFIX cogatlas: <https://www.cognitiveatlas.org/task/id/>
 PREFIX nb: <http://neurobagel.org/vocab/>
@@ -47,6 +48,15 @@
 PREFIX snomed: <http://purl.bioontology.org/ontology/SNOMEDCT/>
 """
 
+CONTEXT = {
+    "cogatlas": "https://www.cognitiveatlas.org/task/id/",
+    "nb": "http://neurobagel.org/vocab/",
+    "nbg": "http://neurobagel.org/graph/",  # TODO: Check if we still need this namespace.
+    "ncit": "http://ncicb.nci.nih.gov/xml/owl/EVS/Thesaurus.owl#",
+    "nidm": "http://purl.org/nidash/nidm#",
+    "snomed": "http://purl.bioontology.org/ontology/SNOMEDCT/",
+}
+
 # Store domains in named tuples
 Domain = namedtuple("Domain", ["var", "pred"])
 # Core domains
@@ -219,6 +229,7 @@ def create_terms_query(data_element_URI: str) -> str:
     -------
     str
         The SPARQL query.
+
     Examples
     --------
     get_terms_query("nb:Assessment")
@@ -227,8 +238,31 @@ def create_terms_query(data_element_URI: str) -> str:
     query_string = f"""
     SELECT DISTINCT ?termURL
     WHERE {{
-        ?termURL a {data_element_URI}.
+        ?termURL a {data_element_URI} .
+        {data_element_URI} rdfs:subClassOf nb:ControlledTerm .
     }}
     """
 
     return "\n".join([DEFAULT_CONTEXT, query_string])
+
+
+def replace_namespace_uri(url: str) -> str:
+    """
+    Replaces namespace URIs in term URLs with corresponding prefixes from the context.
+
+    Parameters
+    ----------
+    url : str
+        A controlled term URL.
+
+    Returns
+    -------
+    str
+        The term with namespace URIs replaced with prefixes if found in the context, or the original URL.
+    """
+    for prefix, uri in CONTEXT.items():
+        if uri in url:
+            return url.replace(uri, f"{prefix}:")
+
+    # If no match found within the context, return original URL
+    return url

app/main.py

@@ -9,7 +9,7 @@
 from fastapi.responses import ORJSONResponse
 
 from .api import utility as util
-from .api.routers import query
+from .api.routers import attributes, query
 
 app = FastAPI(default_response_class=ORJSONResponse)
 
@@ -48,6 +48,7 @@ async def allowed_origins_check():
 
 
 app.include_router(query.router)
+app.include_router(attributes.router)
 
 # Automatically start uvicorn server on execution of main.py
 if __name__ == "__main__":

tests/conftest.py

@@ -106,11 +106,11 @@ def terms_test_data():
     """Create toy data for terms for testing."""
     return {
         "nb:NeurobagelClass": [
-            "http://neurobagel.org/vocab/term1",
-            "http://neurobagel.org/vocab/term2",
-            "http://neurobagel.org/vocab/term3",
-            "http://neurobagel.org/vocab/term4",
-            "http://neurobagel.org/vocab/term5",
+            "nb:term1",
+            "nb:term2",
+            "nb:term3",
+            "nb:term4",
+            "nb:term5",
         ]
     }
 

tests/test_query.py

@@ -406,7 +406,7 @@ def test_get_terms_valid_data_element_URI(
     """Given a valid data element URI, returns a 200 status code and a non-empty list of terms for that data element."""
 
     monkeypatch.setattr(crud, "get_terms", mock_successful_get_terms)
-    response = test_app.get(f"/query/attributes/{valid_data_element_URI}")
+    response = test_app.get(f"/attributes/{valid_data_element_URI}")
     assert response.status_code == 200
     first_key = next(iter(response.json()))
     assert response.json()[first_key] != []
@@ -417,9 +417,57 @@ def test_get_terms_valid_data_element_URI(
     ["apple", "some_thing:cool"],
 )
 def test_get_terms_invalid_data_element_URI(
-    test_app, invalid_data_element_URI, monkeypatch
+    test_app, invalid_data_element_URI
 ):
     """Given an invalid data element URI, returns a 422 status code as the validation of the data element URI fails."""
 
-    response = test_app.get(f"/query/attributes/{invalid_data_element_URI}")
+    response = test_app.get(f"/attributes/{invalid_data_element_URI}")
     assert response.status_code == 422
+
+
+def test_get_attributes(
+    test_app,
+    monkeypatch,
+):
+    """Given a GET request to the /attributes/ endpoint, successfully returns controlled term attributes with namespaces abbrieviated and as a list."""
+
+    monkeypatch.setenv(util.GRAPH_USERNAME.name, "SomeUser")
+    monkeypatch.setenv(util.GRAPH_PASSWORD.name, "SomePassword")
+
+    mock_response_json = {
+        "head": {"vars": ["attribute"]},
+        "results": {
+            "bindings": [
+                {
+                    "attribute": {
+                        "type": "uri",
+                        "value": "http://neurobagel.org/vocab/ControlledTerm1",
+                    }
+                },
+                {
+                    "attribute": {
+                        "type": "uri",
+                        "value": "http://neurobagel.org/vocab/ControlledTerm2",
+                    }
+                },
+                {
+                    "attribute": {
+                        "type": "uri",
+                        "value": "http://neurobagel.org/vocab/ControlledTerm3",
+                    }
+                },
+            ]
+        },
+    }
+
+    def mock_httpx_post(**kwargs):
+        return httpx.Response(status_code=200, json=mock_response_json)
+
+    monkeypatch.setattr(httpx, "post", mock_httpx_post)
+    response = test_app.get("/attributes/")
+
+    assert response.json() == [
+        "nb:ControlledTerm1",
+        "nb:ControlledTerm2",
+        "nb:ControlledTerm3",
+    ]