Provide assets by asset type alongside easier parsing (#523)

Author: PGijsbers

src/routers/helper_functions.py

@@ -3,19 +3,22 @@
 from database.model.helper_functions import non_abstract_subclasses
 
 
-def get_all_asset_schemas():
+def get_all_read_classes() -> dict[str, AIoDConcept]:
+    """Returns a list of all schema types and a reference to their definition."""
     available_schemas: list[AIoDConcept] = list(non_abstract_subclasses(AIoDConcept))
     classes_dict = {clz.__tablename__: clz for clz in available_schemas if clz.__tablename__}
     resrouters = {
         route.resource_name: route
         for route in routers.resource_routers.router_list  # type: ignore
     }
-    read_classes_dict = {
+    return {
         name: resrouters[name].resource_class_read
         for name in classes_dict
         if name not in ["testresource", "test_object"]
     }
-    responses = [
-        {"$ref": f"#/components/schemas/{clz.__name__}"} for clz in read_classes_dict.values()
+
+
+def get_all_asset_schemas():
+    return [
+        {"$ref": f"#/components/schemas/{clz.__name__}"} for clz in get_all_read_classes().values()
     ]
-    return responses

src/routers/user_router.py

@@ -1,6 +1,8 @@
 from http import HTTPStatus
+from typing import List
 
 from fastapi import APIRouter, Depends
+from pydantic import BaseModel, create_model, Field
 from sqlalchemy import select
 from sqlmodel import Session
 
@@ -10,13 +12,24 @@
 from database.model.concept.aiod_entry import AIoDEntryORM
 from database.model.concept.concept import AIoDConcept
 from database.model.helper_functions import non_abstract_subclasses
-from routers.helper_functions import get_all_asset_schemas
+from routers.helper_functions import get_all_read_classes
 
 
 def create(url_prefix: str) -> APIRouter:
     router = APIRouter()
     version = "v1"
 
+    # We define a custom response class here to ensure all the asset
+    # types are included, and the (schema) documentation is generated.
+    # It also makes sure assets are deserialized the same way as
+    # direct access would have.
+    Catalogue = create_model(
+        "Catalogue",
+        **{
+            asset_type: (List[asset_read_class], Field())  # type: ignore[valid-type]
+            for asset_type, asset_read_class in get_all_read_classes().items()
+        },
+    )
     for path in [
         f"{url_prefix}/user/resources/{version}",
         f"{url_prefix}/v2/user/resources",
@@ -26,32 +39,19 @@ def create(url_prefix: str) -> APIRouter:
             path,
             tags=["User"],
             description="Return all assets for which you have administrator rights",
-            response_model=None,  # Required! Otherwise FastAPI infers it from type annotation.
-            responses={
-                HTTPStatus.OK: {
-                    "content": {
-                        "application/json": {
-                            "schema": {
-                                "title": "List of assets owned by the user.",
-                                "type": "array",
-                                "items": {"anyOf": get_all_asset_schemas()},
-                            }
-                        }
-                    },
-                }
-            },
+            response_model=Catalogue,
         )(get_resources_for_logged_in_user)
     return router
 
 
 def get_resources_for_logged_in_user(
     user: KeycloakUser = Depends(get_user_or_raise),
     session: Session = Depends(get_session),
-) -> list[AIoDConcept]:
+) -> dict[str, list[AIoDConcept]]:
     return _get_resources_for_user(user, session)
 
 
-def _get_resources_for_user(user: KeycloakUser, session: Session) -> list[AIoDConcept]:
+def _get_resources_for_user(user: KeycloakUser, session: Session) -> dict[str, list[AIoDConcept]]:
     # "Ownership" is currently equivalent to having ADMIN permissions
     stmt = (
         select(AIoDEntryORM)
@@ -66,15 +66,16 @@ def _get_resources_for_user(user: KeycloakUser, session: Session) -> list[AIoDCo
     # We have AIoD entries, but want their respective asset information (e.g. publication).
     # We lack the information about what the type of the asset is, so unfortunately we
     # have to check all tables:
-    found_assets = []
-    for asset_type in non_abstract_subclasses(AIoDConcept):
-        query = select(asset_type).where(asset_type.aiod_entry_identifier.in_(assets_to_fetch))
+    asset_types = list(non_abstract_subclasses(AIoDConcept))
+    found_assets: dict[str, list[AIoDConcept]] = {type_.__tablename__: [] for type_ in asset_types}
+    for asset_type in asset_types:
+        query = (
+            select(asset_type)
+            .where(asset_type.aiod_entry_identifier.in_(assets_to_fetch))
+            .where(asset_type.date_deleted.is_(None))
+        )
         assets = session.scalars(query).all()
-        found_assets.extend(assets)
-        if len(found_assets) == len(assets_to_fetch):
-            return found_assets  # minor optimization since queries may be expensive
-
-    raise RuntimeError(
-        f"Expected to find assets for identifiers {assets_to_fetch}, "
-        f"but only found {len(found_assets)} in total: {found_assets}."
-    )
+        found_assets[asset_type.__tablename__] = list(assets)
+        if sum(map(len, found_assets.values())) == len(assets_to_fetch):
+            break
+    return found_assets  # minor optimization since queries may be expensive

src/tests/test_user_endpoints.py

@@ -16,23 +16,27 @@ def test_my_resources_can_be_empty(client: TestClient) -> None:
     with logged_in_user(ALICE):
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
     assert response.status_code == HTTPStatus.OK
-    assert response.json() == [], "A user with no resources should get an empty list"
+    msg = "A user with no resources should get an empty list"
+    assert all(items == [] for items in response.json().values()), msg
+    assert len(response.json()) >= 16, "Every asset type should always be included in the response"
 
 
 def test_my_resources_shows_draft_assets(client: TestClient, publication: Publication) -> None:
     register_asset(publication, owner=ALICE, status=EntryStatus.DRAFT)
     with logged_in_user(ALICE):
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
     assert response.status_code == HTTPStatus.OK
-    assert len(response.json()) == 1, "Draft assets should be included in this view."
+    assert len(response.json()["publication"]) == 1, "Draft assets should be included in this view."
+    msg = "Properties should be deserialized"
+    assert response.json()["publication"][0]["aiod_entry"]["status"] == "draft", msg
 
 
 def test_my_resources_shows_published_assets(client: TestClient, publication: Publication) -> None:
     register_asset(publication, owner=ALICE, status=EntryStatus.PUBLISHED)
     with logged_in_user(ALICE):
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
     assert response.status_code == HTTPStatus.OK
-    assert len(response.json()) == 1, "Published assets should be included in this view."
+    assert len(response.json()["publication"]) == 1, "Published assets should be included in this view."
 
 
 def test_my_resources_shows_mixed_assets(client: TestClient, publication: Publication, organisation: Organisation) -> None:
@@ -42,26 +46,32 @@ def test_my_resources_shows_mixed_assets(client: TestClient, publication: Public
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
     assert response.status_code == HTTPStatus.OK
 
-    pub = next(asset for asset in response.json() if asset["aiod_entry_identifier"] == 1)
-    org = next(asset for asset in response.json() if asset["aiod_entry_identifier"] == 2)
+    assert len(response.json()["contact"]) == 0
+    assert len(response.json()["publication"]) == 1
+    assert len(response.json()["organisation"]) == 1
+
+    pub = response.json()["publication"][0]
+    org = response.json()["organisation"][0]
+
     dataset_property = "legal_name"
     publication_property = "isbn"
 
     assert dataset_property in org and publication_property in pub, "Assets should report properties unique to their type"
     assert dataset_property not in pub and publication_property not in org, "Assets should not report properties they do not have"
 
+
 def test_my_resources_shows_only_own_resources(client: TestClient, publication_factory: Callable[[], Publication]) -> None:
     register_asset(publication_factory(), owner=ALICE, status=EntryStatus.DRAFT)
     register_asset(publication_factory(), owner=ALICE, status=EntryStatus.PUBLISHED)
     register_asset(publication_factory(), owner=BOB, status=EntryStatus.PUBLISHED)
 
     with logged_in_user(ALICE):
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
-        assert len(response.json()) == 2
+        assert len(response.json()["publication"]) == 2
 
     with logged_in_user(BOB):
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
-        assert len(response.json()) == 1
+        assert len(response.json()["publication"]) == 1
 
 
 def test_my_resources_counts_only_if_admin(client: TestClient, publication_factory: Callable[[], Publication]) -> None:
@@ -81,8 +91,8 @@ def test_my_resources_counts_only_if_admin(client: TestClient, publication_facto
 
     with logged_in_user(BOB):
         response = client.get("/user/resources", headers={"Authorization": "fake token"})
-        assert len(response.json()) == 1, "Bob has ADMIN permission to one asset."
-        assert response.json()[0]["identifier"] == identifier_three
+        assert len(response.json()["publication"]) == 1, "Bob has ADMIN permission to one asset."
+        assert response.json()["publication"][0]["identifier"] == identifier_three
 
 
 def test_my_resources_must_be_authorized(client: TestClient) -> None: