Add read endpoints for service admins (#354)

Author: eleftherioszisis

app/repository/asset.py

@@ -18,6 +18,8 @@ def get_entity_asset(
         entity_type: EntityType,
         entity_id: uuid.UUID,
         asset_id: uuid.UUID,
+        *,
+        include_deleted: bool = False,
     ) -> Asset:
         """Return a single asset, or raise an error."""
         query = (
@@ -26,10 +28,13 @@ def get_entity_asset(
             .where(
                 Asset.entity_id == entity_id,
                 Asset.id == asset_id,
-                Asset.status != AssetStatus.DELETED,
                 Entity.type == entity_type.name,
             )
         )
+        # See: https://github.com/openbraininstitute/entitycore/issues/358
+        if not include_deleted:
+            query = query.where(Asset.status != AssetStatus.DELETED)
+
         return self.db.execute(query).scalar_one()
 
     def create_entity_asset(self, entity_id: uuid.UUID, asset: AssetCreate) -> Asset:

app/routers/admin.py

@@ -3,13 +3,16 @@
 from fastapi import APIRouter
 
 from app.db.utils import RESOURCE_TYPE_TO_CLASS
+from app.dependencies.common import PaginationQuery
 from app.dependencies.db import RepoGroupDep, SessionDep
 from app.dependencies.s3 import StorageClientFactoryDep
+from app.filters.asset import AssetFilterDep
 from app.queries.common import router_delete_one
 from app.schemas.asset import (
     AssetRead,
 )
-from app.service import asset as asset_service
+from app.schemas.types import ListResponse
+from app.service import admin as admin_service, asset as asset_service
 from app.utils.routers import EntityRoute, ResourceRoute, entity_route_to_type, route_to_type
 
 router = APIRouter(
@@ -33,6 +36,39 @@ def delete_one(
     )
 
 
+@router.get("/{entity_route}/{entity_id}/assets/{asset_id}")
+def get_entity_asset(
+    repos: RepoGroupDep,
+    entity_route: EntityRoute,
+    entity_id: uuid.UUID,
+    asset_id: uuid.UUID,
+) -> AssetRead:
+    """Return an asset associated with a specific entity."""
+    return admin_service.get_entity_asset(
+        repos=repos,
+        entity_type=entity_route_to_type(entity_route),
+        entity_id=entity_id,
+        asset_id=asset_id,
+    )
+
+
+@router.get("/{entity_route}/{entity_id}/assets")
+def get_entity_assets(
+    repos: RepoGroupDep,
+    entity_route: EntityRoute,
+    entity_id: uuid.UUID,
+    pagination_request: PaginationQuery,
+    filter_model: AssetFilterDep,
+) -> ListResponse[AssetRead]:
+    return admin_service.get_entity_assets(
+        repos=repos,
+        entity_route=entity_route,
+        entity_id=entity_id,
+        pagination_request=pagination_request,
+        filter_model=filter_model,
+    )
+
+
 @router.delete("/{entity_route}/{entity_id}/assets/{asset_id}")
 def delete_entity_asset(
     repos: RepoGroupDep,

app/routers/brain_atlas.py

@@ -1,10 +1,13 @@
 from fastapi import APIRouter
 
 import app.service.brain_atlas
+from app.routers.admin import router as admin_router
+
+ROUTE = "brain-atlas"
 
 router = APIRouter(
-    prefix="/brain-atlas",
-    tags=["brain-atlas"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_many = router.get("")(app.service.brain_atlas.read_many)
@@ -13,3 +16,5 @@
 read_one_region = router.get("/{atlas_id}/regions/{atlas_region_id}")(
     app.service.brain_atlas.read_one_region
 )
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{atlas_id}}")(app.service.brain_atlas.admin_read_one)

app/routers/calibration.py

@@ -1,14 +1,19 @@
 from fastapi import APIRouter
 
 import app.service.calibration
+from app.routers.admin import router as admin_router
+
+ROUTE = "calibration"
 
 router = APIRouter(
-    prefix="/calibration",
-    tags=["calibration"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_many = router.get("")(app.service.calibration.read_many)
 read_one = router.get("/{id_}")(app.service.calibration.read_one)
 create_one = router.post("")(app.service.calibration.create_one)
 delete_one = router.delete("/{id_}")(app.service.calibration.delete_one)
 update_one = router.patch("/{id_}")(app.service.calibration.update_one)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(app.service.calibration.admin_read_one)

app/routers/cell_composition.py

@@ -1,11 +1,16 @@
 from fastapi import APIRouter
 
 import app.service.cell_composition
+from app.routers.admin import router as admin_router
+
+ROUTE = "cell-composition"
 
 router = APIRouter(
-    prefix="/cell-composition",
-    tags=["cell-composition"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_one = router.get("/{id_}")(app.service.cell_composition.read_one)
 read_many = router.get("")(app.service.cell_composition.read_many)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(app.service.cell_composition.admin_read_one)

app/routers/circuit.py

@@ -2,14 +2,19 @@
 
 import app.service.circuit
 import app.service.hierarchy
+from app.routers.admin import router as admin_router
+
+ROUTE = "circuit"
 
 router = APIRouter(
-    prefix="/circuit",
-    tags=["circuit"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 hierarchy = router.get("/hierarchy")(app.service.hierarchy.read_circuit_hierarchy)
 read_many = router.get("")(app.service.circuit.read_many)
 read_one = router.get("/{id_}")(app.service.circuit.read_one)
 create_one = router.post("")(app.service.circuit.create_one)
 update_one = router.patch("/{id_}")(app.service.circuit.update_one)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(app.service.circuit.admin_read_one)

app/routers/consortium.py

@@ -1,12 +1,17 @@
 from fastapi import APIRouter
 
 import app.service.consortium
+from app.routers.admin import router as admin_router
+
+ROUTE = "consortium"
 
 router = APIRouter(
-    prefix="/consortium",
-    tags=["consortium"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_many = router.get("")(app.service.consortium.read_many)
 read_one = router.get("/{id_}")(app.service.consortium.read_one)
 create_one = router.post("")(app.service.consortium.create_one)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(app.service.consortium.admin_read_one)

app/routers/contribution.py

@@ -1,12 +1,17 @@
 from fastapi import APIRouter
 
 import app.service.contribution
+from app.routers.admin import router as admin_router
+
+ROUTE = "contribution"
 
 router = APIRouter(
-    prefix="/contribution",
-    tags=["contribution"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_many = router.get("")(app.service.contribution.read_many)
 read_one = router.get("/{id_}")(app.service.contribution.read_one)
 create_one = router.post("")(app.service.contribution.create_one)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(app.service.contribution.admin_read_one)

app/routers/electrical_cell_recording.py

@@ -1,13 +1,20 @@
 from fastapi import APIRouter
 
 import app.service.electrical_cell_recording
+from app.routers.admin import router as admin_router
+
+ROUTE = "electrical-cell-recording"
 
 router = APIRouter(
-    prefix="/electrical-cell-recording",
-    tags=["electrical-cell-recording"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_many = router.get("")(app.service.electrical_cell_recording.read_many)
 read_one = router.get("/{id_}")(app.service.electrical_cell_recording.read_one)
 create_one = router.post("")(app.service.electrical_cell_recording.create_one)
 update_one = router.patch("/{id_}")(app.service.electrical_cell_recording.update_one)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(
+    app.service.electrical_cell_recording.admin_read_one
+)

app/routers/electrical_recording_stimulus.py

@@ -1,13 +1,20 @@
 from fastapi import APIRouter
 
 import app.service.electrical_recording_stimulus
+from app.routers.admin import router as admin_router
+
+ROUTE = "electrical-recording-stimulus"
 
 router = APIRouter(
-    prefix="/electrical-recording-stimulus",
-    tags=["electrical-recording-stimulus"],
+    prefix=f"/{ROUTE}",
+    tags=[ROUTE],
 )
 
 read_many = router.get("")(app.service.electrical_recording_stimulus.read_many)
 read_one = router.get("/{id_}")(app.service.electrical_recording_stimulus.read_one)
 create_one = router.post("")(app.service.electrical_recording_stimulus.create_one)
 update_one = router.patch("/{id_}")(app.service.electrical_recording_stimulus.update_one)
+
+admin_read_one = admin_router.get(f"/{ROUTE}/{{id_}}")(
+    app.service.electrical_recording_stimulus.admin_read_one
+)