Example admin update endpoint

Author: eleftherioszisis

app/queries/common.py

@@ -334,15 +334,17 @@ def router_update_one[T: BaseModel, I: Identifiable](
     id_: uuid.UUID,
     db: Session,
     db_model_class: type[I],
-    user_context: UserContext,
+    user_context: UserContext | None,
     json_model: BaseModel,
     response_schema_class: type[T],
     apply_operations: ApplyOperations | None = None,
 ):
     query = (
         sa.select(db_model_class).where(db_model_class.id == id_).with_for_update(of=db_model_class)
     )
-    if id_model_class := get_declaring_class(db_model_class, "authorized_project_id"):
+    if user_context and (
+        id_model_class := get_declaring_class(db_model_class, "authorized_project_id")
+    ):
         query = constrain_to_private_entities(query, user_context, db_model_class=id_model_class)
     if apply_operations:
         query = apply_operations(query)

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_update_one = admin_router.patch(f"/{ROUTE}/{{id_}}")(
+    app.service.electrical_cell_recording.admin_update_one
+)

app/schemas/electrical_cell_recording.py

@@ -72,8 +72,15 @@ class ElectricalCellRecordingCreate(ElectricalCellRecordingBase, ScientificArtif
     pass
 
 
-ElectricalCellRecordingUpdate = make_update_schema(
-    ElectricalCellRecordingCreate, "ElectricalCellRecordingUpdate"
+ElectricalCellRecordingUserUpdate = make_update_schema(
+    ElectricalCellRecordingCreate,
+    "ElectricalCellRecordingUserUpdate",
+)  # pyright : ignore [reportInvalidTypeForm]
+
+ElectricalCellRecordingAdminUpdate = make_update_schema(
+    ElectricalCellRecordingCreate,
+    "ElectricalCellRecordingAdminUpdate",
+    excluded_fields=set(),
 )  # pyright : ignore [reportInvalidTypeForm]
 
 

app/schemas/utils.py

@@ -8,7 +8,11 @@
 }
 
 
-def make_update_schema(schema: type[BaseModel], new_schema_name: str):
+def make_update_schema(
+    schema: type[BaseModel],
+    new_schema_name: str | None = None,
+    excluded_fields: set = EXCLUDED_FIELDS,
+):
     """Create a new pydantic schema from current schema where all fields are optional.
 
     In order to differentiate between the user providing a None value and an actual not set by the
@@ -27,6 +31,6 @@ def make_optional(field):
     fields = {
         name: make_optional(field)
         for name, field in schema.model_fields.items()
-        if name not in EXCLUDED_FIELDS
+        if name not in excluded_fields
     }
     return create_model(new_schema_name, **fields)  # pyright: ignore reportArgumentType

app/service/electrical_cell_recording.py

@@ -27,9 +27,10 @@
 )
 from app.queries.factory import query_params_factory
 from app.schemas.electrical_cell_recording import (
+    ElectricalCellRecordingAdminUpdate,
     ElectricalCellRecordingCreate,
     ElectricalCellRecordingRead,
-    ElectricalCellRecordingUpdate,
+    ElectricalCellRecordingUserUpdate,
 )
 from app.schemas.types import ListResponse
 
@@ -153,7 +154,7 @@ def update_one(
     user_context: UserContextDep,
     db: SessionDep,
     id_: uuid.UUID,
-    json_model: ElectricalCellRecordingUpdate,  # pyright: ignore [reportInvalidTypeForm]
+    json_model: ElectricalCellRecordingUserUpdate,  # pyright: ignore [reportInvalidTypeForm]
 ) -> ElectricalCellRecordingRead:
     return router_update_one(
         id_=id_,
@@ -164,3 +165,19 @@ def update_one(
         response_schema_class=ElectricalCellRecordingRead,
         apply_operations=_load,
     )
+
+
+def admin_update_one(
+    db: SessionDep,
+    id_: uuid.UUID,
+    json_model: ElectricalCellRecordingAdminUpdate,  # pyright: ignore [reportInvalidTypeForm]
+) -> ElectricalCellRecordingRead:
+    return router_update_one(
+        id_=id_,
+        db=db,
+        db_model_class=ElectricalCellRecording,
+        user_context=None,
+        json_model=json_model,
+        response_schema_class=ElectricalCellRecordingRead,
+        apply_operations=_load,
+    )

tests/test_electrical_cell_recording.py

@@ -56,7 +56,7 @@ def test_create_one(
     assert data["etypes"] == []
 
 
-def test_update_one(client, trace_id_with_assets):
+def test_update_one(client, client_admin, trace_id_with_assets):
     new_name = "my_new_name"
     new_description = "my_new_description"
 
@@ -92,8 +92,64 @@ def test_update_one(client, trace_id_with_assets):
     ).json()
     assert data["temperature"] is None
 
+    # only admin client can hit admin endpoint
+    data = assert_request(
+        client.patch,
+        url=f"{ADMIN_ROUTE}/{trace_id_with_assets}",
+        json={
+            "name": new_name,
+            "description": new_description,
+        },
+        expected_status_code=403,
+    ).json()
+    assert data["error_code"] == "NOT_AUTHORIZED"
+    assert data["message"] == "Service admin role required"
+
+    data = assert_request(
+        client_admin.patch,
+        url=f"{ADMIN_ROUTE}/{trace_id_with_assets}",
+        json={
+            "name": new_name,
+            "description": new_description,
+        },
+    ).json()
 
-def test_update_one__public(client, electrical_cell_recording_json_data):
+    assert data["name"] == new_name
+    assert data["description"] == new_description
+
+    # set temperature
+    data = assert_request(
+        client_admin.patch,
+        url=f"{ADMIN_ROUTE}/{trace_id_with_assets}",
+        json={
+            "temperature": 10.0,
+        },
+    ).json()
+    assert data["temperature"] == 10.0
+
+    # unset temperature
+    data = assert_request(
+        client_admin.patch,
+        url=f"{ADMIN_ROUTE}/{trace_id_with_assets}",
+        json={
+            "temperature": None,
+        },
+    ).json()
+    assert data["temperature"] is None
+
+    # admin is treated as regular user for regular route (no authorized project ids)
+    data = assert_request(
+        client_admin.patch,
+        url=f"{ROUTE}/{trace_id_with_assets}",
+        json={
+            "temperature": None,
+        },
+        expected_status_code=404,
+    ).json()
+    assert data["error_code"] == "ENTITY_NOT_FOUND"
+
+
+def test_user_update_one__public(client, client_admin, electrical_cell_recording_json_data):
     # make private entity public
     data = assert_request(
         client.post,
@@ -104,15 +160,29 @@ def test_update_one__public(client, electrical_cell_recording_json_data):
         },
     ).json()
 
+    entity_id = data["id"]
+
     # should not be allowed to update it once public
     data = assert_request(
         client.patch,
-        url=f"{ROUTE}/{data['id']}",
+        url=f"{ROUTE}/{entity_id}",
         json={"name": "foo"},
         expected_status_code=404,
     ).json()
     assert data["error_code"] == "ENTITY_NOT_FOUND"
 
+    # admin has no such restrictions
+    data = assert_request(
+        client_admin.patch,
+        url=f"{ADMIN_ROUTE}/{entity_id}",
+        json={
+            "authorized_public": False,
+            "name": "foo",
+        },
+    ).json()
+    assert data["authorized_public"] is False
+    assert data["name"] == "foo"
+
 
 def test_read_one(client, subject_id, license_id, brain_region_id, trace_id_with_assets):
     data = assert_request(