allow morphology get by id to retrieve by legacy id

.gitignore

@@ -12,3 +12,4 @@ out.tar
 .DS_Store
 mba_hierarchy.json
 notes.md
+.env.local

app/routers/morphology.py

@@ -1,8 +1,10 @@
+import urllib.parse
 import uuid
+from http import HTTPStatus
 from typing import Annotated
 
 import sqlalchemy as sa
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, Depends, Path, Query
 from fastapi_filter import FilterDepends
 from sqlalchemy.orm import (
     aliased,
@@ -26,7 +28,7 @@
 from app.dependencies.auth import UserContextDep, UserContextWithProjectIdDep
 from app.dependencies.common import PaginationQuery
 from app.dependencies.db import SessionDep
-from app.errors import ensure_result
+from app.errors import ApiError, ApiErrorCode, ensure_result
 from app.filters.morphology import MorphologyFilter
 from app.routers.common import FacetQueryParams, _get_facets
 from app.schemas.morphology import (
@@ -42,55 +44,6 @@
 )
 
 
-@router.get(
-    "/{id_}",
-    response_model=ReconstructionMorphologyRead | ReconstructionMorphologyAnnotationExpandedRead,
-)
-def read_reconstruction_morphology(
-    user_context: UserContextDep,
-    db: SessionDep,
-    id_: uuid.UUID,
-    expand: Annotated[set[str] | None, Query()] = None,
-):
-    with ensure_result(error_message="ReconstructionMorphology not found"):
-        query = constrain_to_accessible_entities(
-            sa.select(ReconstructionMorphology), user_context.project_id
-        ).filter(ReconstructionMorphology.id == id_)
-
-        if expand and "morphology_feature_annotation" in expand:
-            query = query.options(
-                joinedload(ReconstructionMorphology.morphology_feature_annotation)
-                .selectinload(MorphologyFeatureAnnotation.measurements)
-                .selectinload(MorphologyMeasurement.measurement_serie)
-            )
-
-        query = (
-            query.options(joinedload(ReconstructionMorphology.brain_region))
-            .options(
-                selectinload(ReconstructionMorphology.contributions).selectinload(
-                    Contribution.agent
-                )
-            )
-            .options(
-                selectinload(ReconstructionMorphology.contributions).selectinload(Contribution.role)
-            )
-            .options(joinedload(ReconstructionMorphology.mtypes))
-            .options(joinedload(ReconstructionMorphology.license))
-            .options(joinedload(ReconstructionMorphology.species))
-            .options(joinedload(ReconstructionMorphology.strain))
-            .options(selectinload(ReconstructionMorphology.assets))
-            .options(raiseload("*"))
-        )
-
-        row = db.execute(query).unique().scalar_one()
-
-    if expand and "morphology_feature_annotation" in expand:
-        return ReconstructionMorphologyAnnotationExpandedRead.model_validate(row)
-
-    # added back with None by the response_model
-    return ReconstructionMorphologyRead.model_validate(row)
-
-
 @router.post("", response_model=ReconstructionMorphologyRead)
 def create_reconstruction_morphology(
     user_context: UserContextWithProjectIdDep,
@@ -213,3 +166,92 @@ def morphology_query(
     )
 
     return response
+
+
+def validate_id(
+    id_: str = Path(...),
+    *,
+    is_legacy: Annotated[bool, Query()] = False,
+) -> tuple[str, bool]:
+    if not is_legacy:
+        try:
+            uuid.UUID(id_)
+        except ValueError as err:
+            raise ApiError(
+                message=f"Invalid UUID format: {id_}",
+                error_code=ApiErrorCode.INVALID_REQUEST,
+                http_status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+            ) from err
+    else:
+        try:
+            decoded_id = urllib.parse.unquote(id_)
+            parsed = urllib.parse.urlparse(decoded_id)
+            if not parsed.scheme or not parsed.netloc:
+                msg = "Missing scheme or host in URL"
+                raise ValueError(msg)  # noqa: TRY301
+        except ValueError as err:
+            raise ApiError(
+                message=f"Invalid URL format for legacy ID: {id_}",
+                error_code=ApiErrorCode.INVALID_REQUEST,
+                http_status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+            ) from err
+
+    return id_, is_legacy
+
+
+@router.get(
+    "/{id_:path}",
+    response_model=ReconstructionMorphologyRead | ReconstructionMorphologyAnnotationExpandedRead,
+)
+def read_reconstruction_morphology(
+    user_context: UserContextDep,
+    db: SessionDep,
+    validated_id: Annotated[tuple[str, bool], Depends(validate_id)],
+    expand: Annotated[set[str] | None, Query()] = None,
+):
+    id_, is_legacy = validated_id
+
+    with ensure_result(error_message="ReconstructionMorphology not found"):
+        if is_legacy:
+            decoded_id = urllib.parse.unquote(id_)
+            query = constrain_to_accessible_entities(
+                sa.select(ReconstructionMorphology), user_context.project_id
+            ).filter(decoded_id == sa.any_(ReconstructionMorphology.legacy_id))
+        else:
+            uuid_id = uuid.UUID(id_)
+            query = constrain_to_accessible_entities(
+                sa.select(ReconstructionMorphology), user_context.project_id
+            ).filter(ReconstructionMorphology.id == uuid_id)
+
+        if expand and "morphology_feature_annotation" in expand:
+            query = query.options(
+                joinedload(ReconstructionMorphology.morphology_feature_annotation)
+                .selectinload(MorphologyFeatureAnnotation.measurements)
+                .selectinload(MorphologyMeasurement.measurement_serie)
+            )
+
+        query = (
+            query.options(joinedload(ReconstructionMorphology.brain_region))
+            .options(
+                selectinload(ReconstructionMorphology.contributions).selectinload(
+                    Contribution.agent
+                )
+            )
+            .options(
+                selectinload(ReconstructionMorphology.contributions).selectinload(Contribution.role)
+            )
+            .options(joinedload(ReconstructionMorphology.mtypes))
+            .options(joinedload(ReconstructionMorphology.license))
+            .options(joinedload(ReconstructionMorphology.species))
+            .options(joinedload(ReconstructionMorphology.strain))
+            .options(selectinload(ReconstructionMorphology.assets))
+            .options(raiseload("*"))
+        )
+
+        row = db.execute(query).unique().scalar_one()
+
+    if expand and "morphology_feature_annotation" in expand:
+        return ReconstructionMorphologyAnnotationExpandedRead.model_validate(row)
+
+    # added back with None by the response_model
+    return ReconstructionMorphologyRead.model_validate(row)

app/schemas/morphology.py

@@ -66,6 +66,7 @@ class ReconstructionMorphologyRead(
     contributions: list[ContributionReadWithoutEntity] | None
     mtypes: list[MTypeClassRead] | None
     assets: list[AssetRead] | None
+    type: str = "reconstruction_morphology"
 
 
 class ReconstructionMorphologyAnnotationExpandedRead(ReconstructionMorphologyRead):

tests/test_morphology.py

@@ -1,4 +1,5 @@
 import itertools as it
+import urllib.parse
 
 from app.db.model import ReconstructionMorphology, Species, Strain
 
@@ -341,3 +342,81 @@ def test_pagination(db, client, brain_region_id):
     assert len(items) == total_items
     names = [int(d["name"].removeprefix("TestMorphologyName")) for d in items]
     assert list(reversed(names)) == list(range(total_items))
+
+
+def test_get_morphology_by_uuid_success(client, db, brain_region_id):
+    """Test successful retrieval of a morphology using its UUID."""
+
+    species = add_db(db, Species(name="UUIDTestSpecies", taxonomy_id="0"))
+    strain = add_db(db, Strain(name="UUIDTestStrain", species_id=species.id, taxonomy_id="0"))
+
+    morphology_id = create_reconstruction_morphology_id(
+        client,
+        species_id=species.id,
+        strain_id=strain.id,
+        brain_region_id=brain_region_id,
+        name="UUID Test Morphology",
+        description="Test morphology for UUID access",
+        authorized_public=True,
+    )
+
+    response = client.get(f"{ROUTE}/{morphology_id}", params={"is_legacy": False})
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["id"] == morphology_id
+    assert data["name"] == "UUID Test Morphology"
+    assert data["species"]["name"] == "UUIDTestSpecies"
+
+
+def test_get_morphology_by_legacy_id_success(client, db, brain_region_id):
+    """Test successful retrieval of a morphology using its legacy ID."""
+
+    species = add_db(db, Species(name="LegacyTestSpecies", taxonomy_id="0"))
+    strain = add_db(db, Strain(name="LegacyTestStrain", species_id=species.id, taxonomy_id="0"))
+
+    legacy_id = "https://example.org/morphology/12345"
+
+    # Add a morphology with the legacy ID directly to the database
+    morphology = add_db(
+        db,
+        ReconstructionMorphology(
+            brain_region_id=brain_region_id,
+            species_id=species.id,
+            strain_id=strain.id,
+            name="Legacy ID Test Morphology",
+            description="Test morphology for legacy ID access",
+            authorized_public=True,
+            legacy_id=[legacy_id],
+            authorized_project_id=PROJECT_ID,
+        ),
+    )
+
+    encoded_legacy_id = urllib.parse.quote(legacy_id, safe="")
+    response = client.get(f"{ROUTE}/{encoded_legacy_id}", params={"is_legacy": True})
+
+    assert response.status_code == 200
+    data = response.json()
+    assert str(morphology.id) == data["id"]
+    assert data["name"] == "Legacy ID Test Morphology"
+    assert legacy_id in data["legacy_id"]
+
+
+def test_get_morphology_by_invalid_uuid(client):
+    """Test that an invalid UUID returns a 422 error."""
+    invalid_id = "not-a-valid-uuid"
+    response = client.get(f"{ROUTE}/{invalid_id}")
+
+    assert response.status_code == 422
+    data = response.json()
+    assert "Invalid UUID format" in data["message"]
+
+
+def test_get_morphology_by_invalid_legacy_id(client):
+    """Test that an invalid legacy ID returns a 422 error."""
+    invalid_legacy_id = "invalid-legacy-id-no-url-format"
+    response = client.get(f"{ROUTE}/{invalid_legacy_id}", params={"is_legacy": True})
+
+    assert response.status_code == 422
+    data = response.json()
+    assert "Invalid URL format for legacy ID" in data["message"]