Move query_params dependency logic into QueryParams model

Author: edan-bainglass

aiida_restapi/common/query.py

@@ -6,23 +6,24 @@
 import typing as t
 
 import pydantic as pdt
-from fastapi import HTTPException, Query
 
 
 class QueryParams(pdt.BaseModel):
-    filters: dict[str, t.Any] = pdt.Field(
-        default_factory=dict,
+    filters: dict[str, t.Any] | None = pdt.Field(
+        None,
         description='AiiDA QueryBuilder filters',
         examples=[
-            {'node_type': {'==': 'data.core.int.Int.'}},
-            {'attributes.value': {'>': 42}},
+            '{"node_type": "data.core.int.Int."}',
+            '{"attributes.value": {">": 42}}',
         ],
     )
     order_by: str | list[str] | dict[str, t.Any] | None = pdt.Field(
         None,
         description='Fields to sort by',
         examples=[
-            {'attributes.value': 'desc'},
+            'pk',
+            'uuid,label',
+            '{"attributes.value": "desc"}',
         ],
     )
     page_size: pdt.PositiveInt = pdt.Field(
@@ -36,55 +37,26 @@ class QueryParams(pdt.BaseModel):
         examples=[1],
     )
 
-
-def query_params(
-    filters: str | None = Query(
-        None,
-        description='AiiDA QueryBuilder filters as JSON string',
-    ),
-    order_by: str | None = Query(
-        None,
-        description='Comma-separated list of fields to sort by',
-    ),
-    page_size: pdt.PositiveInt = Query(
-        10,
-        description='Number of results per page',
-    ),
-    page: pdt.PositiveInt = Query(
-        1,
-        description='Page number',
-    ),
-) -> QueryParams:
-    """Parse query parameters into a structured object.
-
-    :param filters: AiiDA QueryBuilder filters as JSON string.
-    :param order_by: Comma-separated string of fields to sort by.
-    :param page_size: Number of results per page.
-    :param page: Page number.
-    :return: Structured query parameters.
-    :raises HTTPException: If filters cannot be parsed as JSON.
-    """
-    query_filters: dict[str, t.Any] = {}
-    query_order_by: str | list[str] | dict[str, t.Any] | None = None
-    if filters:
-        try:
-            query_filters = json.loads(filters)
-        except Exception as exception:
-            raise HTTPException(
-                status_code=400,
-                detail=f'Could not parse filters as JSON: {exception}',
-            ) from exception
-    if order_by:
-        try:
-            query_order_by = json.loads(order_by)
-        except Exception as exception:
-            raise HTTPException(
-                status_code=400,
-                detail=f'Could not parse order_by as JSON: {exception}',
-            ) from exception
-    return QueryParams(
-        filters=query_filters,
-        order_by=query_order_by,
-        page_size=page_size,
-        page=page,
-    )
+    @pdt.field_validator('filters', mode='before')
+    @classmethod
+    def parse_filters(cls, value: t.Any) -> dict[str, t.Any] | None:
+        if value:
+            try:
+                return json.loads(value)
+            except Exception as exception:
+                raise ValueError(f'Could not parse filters as JSON: {exception}') from exception
+        return None
+
+    @pdt.field_validator('order_by', mode='before')
+    @classmethod
+    def parse_order_by(cls, value: t.Any) -> str | list[str] | dict[str, t.Any] | None:
+        if value:
+            # Due to allowing list[str] on the field, FastAPI will always convert query to a list
+            raw: str = value[0]
+            if raw.startswith('{') or raw.startswith('['):
+                try:
+                    return json.loads(raw)
+                except Exception as exception:
+                    raise ValueError(f'Could not parse order_by as JSON: {exception}') from exception
+            return raw.split(',')
+        return None

aiida_restapi/repository/entity.py

@@ -45,23 +45,23 @@ def get_projectable_properties(self) -> list[str]:
         """
         return self.entity_class.fields.keys()
 
-    def get_entities(self, queries: QueryParams) -> PaginatedResults[EntityModelType]:
+    def get_entities(self, query_params: QueryParams) -> PaginatedResults[EntityModelType]:
         """Get AiiDA entities with optional filtering, sorting, and/or pagination.
 
-        :param queries: The query parameters, including filters, order_by, page_size, and page.
+        :param query_params: The query parameters, including filters, order_by, page_size, and page.
         :return: The paginated results, including total count, current page, page size, and list of entity models.
         """
-        total = self.entity_class.collection.count(filters=queries.filters)
+        total = self.entity_class.collection.count(filters=query_params.filters)
         results = self.entity_class.collection.find(
-            filters=queries.filters,
-            order_by=queries.order_by,
-            limit=queries.page_size,
-            offset=queries.page_size * (queries.page - 1),
+            filters=query_params.filters,
+            order_by=query_params.order_by,
+            limit=query_params.page_size,
+            offset=query_params.page_size * (query_params.page - 1),
         )
         return PaginatedResults(
             total=total,
-            page=queries.page,
-            page_size=queries.page_size,
+            page=query_params.page,
+            page_size=query_params.page_size,
             results=[self.to_model(result) for result in results],
         )
 

aiida_restapi/repository/node.py

@@ -148,21 +148,21 @@ def get_node_attributes(self, uuid: str) -> dict[str, t.Any]:
     def get_node_links(
         self,
         uuid: str,
-        queries: QueryParams,
+        query_params: QueryParams,
         direction: t.Literal['incoming', 'outgoing'],
     ) -> PaginatedResults[NodeLinks]:
         """Get the incoming links of a node.
 
         :param uuid: The uuid of the node to retrieve the incoming links for.
-        :param queries: The query parameters, including filters, order_by, page_size, and page.
+        :param query_params: The query parameters, including filters, order_by, page_size, and page.
         :param direction: Specify whether to retrieve incoming or outgoing links.
         :return: The paginated requested linked nodes.
         """
         node = self.entity_class.collection.get(uuid=uuid)
 
         start, end = (
-            queries.page_size * (queries.page - 1),
-            queries.page_size * queries.page,
+            query_params.page_size * (query_params.page - 1),
+            query_params.page_size * query_params.page,
         )
 
         if direction == 'incoming':
@@ -181,8 +181,8 @@ def get_node_links(
 
         return PaginatedResults(
             total=len(links),
-            page=queries.page,
-            page_size=queries.page_size,
+            page=query_params.page,
+            page_size=query_params.page_size,
             results=links,
         )
 

aiida_restapi/routers/computers.py

@@ -10,7 +10,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
 
 from aiida_restapi.common.pagination import PaginatedResults
-from aiida_restapi.common.query import QueryParams, query_params
+from aiida_restapi.common.query import QueryParams
 from aiida_restapi.repository.entity import EntityRepository
 
 from .auth import UserInDB, get_current_active_user
@@ -66,14 +66,27 @@ async def get_computer_projectable_properties() -> list[str]:
 )
 @with_dbenv()
 async def get_computers(
-    queries: t.Annotated[QueryParams, Depends(query_params)],
+    query_params: t.Annotated[
+        QueryParams,
+        Query(
+            default_factory=QueryParams,
+            description='Query parameters for filtering, sorting, and pagination.',
+        ),
+    ],
 ) -> PaginatedResults[orm.Computer.Model]:
     """Get AiiDA computers with optional filtering, sorting, and/or pagination.
 
-    :param queries: The query parameters, including filters, order_by, page_size, and page.
+    :param query_params: The query parameters, including filters, order_by, page_size, and page.
     :return: The paginated results, including total count, current page, page size, and list of computer models.
+    :raises HTTPException: 422 if the query parameters are invalid,
+        500 for other failures during retrieval.
     """
-    return repository.get_entities(queries)
+    try:
+        return repository.get_entities(query_params)
+    except ValueError as exception:
+        raise HTTPException(status_code=422, detail=str(exception)) from exception
+    except Exception as exception:
+        raise HTTPException(status_code=500, detail=str(exception)) from exception
 
 
 @read_router.get(

aiida_restapi/routers/groups.py

@@ -10,7 +10,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
 
 from aiida_restapi.common.pagination import PaginatedResults
-from aiida_restapi.common.query import QueryParams, query_params
+from aiida_restapi.common.query import QueryParams
 from aiida_restapi.repository.entity import EntityRepository
 
 from .auth import UserInDB, get_current_active_user
@@ -66,14 +66,27 @@ async def get_group_projectable_properties() -> list[str]:
 )
 @with_dbenv()
 async def get_groups(
-    queries: t.Annotated[QueryParams, Depends(query_params)],
+    query_params: t.Annotated[
+        QueryParams,
+        Query(
+            default_factory=QueryParams,
+            description='Query parameters for filtering, sorting, and pagination.',
+        ),
+    ],
 ) -> PaginatedResults[orm.Group.Model]:
     """Get AiiDA groups with optional filtering, sorting, and/or pagination.
 
-    :param queries: The query parameters, including filters, order_by, page_size, and page.
+    :param query_params: The query parameters, including filters, order_by, page_size, and page.
     :return: The paginated results, including total count, current page, page size, and list of group models.
+    :raises HTTPException: 422 if the query parameters are invalid,
+        500 for other failures during retrieval.
     """
-    return repository.get_entities(queries)
+    try:
+        return repository.get_entities(query_params)
+    except ValueError as exception:
+        raise HTTPException(status_code=422, detail=str(exception)) from exception
+    except Exception as exception:
+        raise HTTPException(status_code=500, detail=str(exception)) from exception
 
 
 @read_router.get(

aiida_restapi/routers/nodes.py

@@ -15,7 +15,7 @@
 from typing_extensions import TypeAlias
 
 from aiida_restapi.common.pagination import PaginatedResults
-from aiida_restapi.common.query import QueryParams, query_params
+from aiida_restapi.common.query import QueryParams
 from aiida_restapi.config import API_CONFIG
 from aiida_restapi.models.node import NodeModelRegistry
 from aiida_restapi.repository.node import NodeLinks, NodeRepository
@@ -182,14 +182,27 @@ async def get_nodes_download_formats() -> dict[str, t.Any]:
 )
 @with_dbenv()
 async def get_nodes(
-    queries: t.Annotated[QueryParams, Depends(query_params)],
+    query_params: t.Annotated[
+        QueryParams,
+        Query(
+            default_factory=QueryParams,
+            description='Query parameters for filtering, sorting, and pagination.',
+        ),
+    ],
 ) -> PaginatedResults[orm.Node.Model]:
     """Get AiiDA nodes with optional filtering, sorting, and/or pagination.
 
-    :param queries: The query parameters, including filters, order_by, page_size, and page.
+    :param query_params: The query parameters, including filters, order_by, page_size, and page.
     :return: The paginated results, including total count, current page, page size, and list of node models.
+    :raises HTTPException: 422 if the query parameters are invalid,
+        500 for other failures during retrieval.
     """
-    return repository.get_entities(queries)
+    try:
+        return repository.get_entities(query_params)
+    except ValueError as exception:
+        raise HTTPException(status_code=422, detail=str(exception)) from exception
+    except Exception as exception:
+        raise HTTPException(status_code=500, detail=str(exception)) from exception
 
 
 class NodeType(pdt.BaseModel):
@@ -311,22 +324,28 @@ async def get_node_extras(uuid: str) -> dict[str, t.Any]:
 @with_dbenv()
 async def get_node_links(
     uuid: str,
-    queries: t.Annotated[QueryParams, Depends(query_params)],
+    query_params: t.Annotated[
+        QueryParams,
+        Query(
+            default_factory=QueryParams,
+            description='Query parameters for filtering, sorting, and pagination.',
+        ),
+    ],
     direction: t.Literal['incoming', 'outgoing'] = Query(
         description='Specify whether to retrieve incoming or outgoing links.',
     ),
 ) -> PaginatedResults[NodeLinks]:
     """Get the incoming or outgoing links of a node.
 
     :param uuid: The uuid of the node to retrieve the incoming links for.
-    :param queries: The query parameters, including filters, order_by, page_size, and page.
+    :param query_params: The query parameters, including filters, order_by, page_size, and page.
     :param direction: Specify whether to retrieve incoming or outgoing links.
     :return: The paginated requested linked nodes.
     :raises HTTPException: 404 if the node with the given uuid does not exist,
         500 for other failures during retrieval.
     """
     try:
-        return repository.get_node_links(uuid, queries, direction=direction)
+        return repository.get_node_links(uuid, query_params, direction=direction)
     except NotExistent as exception:
         raise HTTPException(status_code=404, detail=str(exception)) from exception
     except Exception as exception:

aiida_restapi/routers/users.py

@@ -10,7 +10,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
 
 from aiida_restapi.common.pagination import PaginatedResults
-from aiida_restapi.common.query import QueryParams, query_params
+from aiida_restapi.common.query import QueryParams
 from aiida_restapi.repository.entity import EntityRepository
 
 from .auth import UserInDB, get_current_active_user
@@ -66,14 +66,27 @@ async def get_user_projectable_properties() -> list[str]:
 )
 @with_dbenv()
 async def get_users(
-    queries: t.Annotated[QueryParams, Depends(query_params)],
+    query_params: t.Annotated[
+        QueryParams,
+        Query(
+            default_factory=QueryParams,
+            description='Query parameters for filtering, sorting, and pagination.',
+        ),
+    ],
 ) -> PaginatedResults[orm.User.Model]:
     """Get AiiDA users with optional filtering, sorting, and/or pagination.
 
-    :param queries: The query parameters, including filters, order_by, page_size, and page.
+    :param query_params: The query parameters, including filters, order_by, page_size, and page.
     :return: The paginated results, including total count, current page, page size, and list of user models.
+    :raises HTTPException: 422 if the query parameters are invalid,
+        500 for other failures during retrieval.
     """
-    return repository.get_entities(queries)
+    try:
+        return repository.get_entities(query_params)
+    except ValueError as exception:
+        raise HTTPException(status_code=422, detail=str(exception)) from exception
+    except Exception as exception:
+        raise HTTPException(status_code=500, detail=str(exception)) from exception
 
 
 @read_router.get(