updated logic to exclude

Author: CosmoV

fastapi_jsonapi/querystring.py

@@ -8,7 +8,6 @@
     List,
     Optional,
     Type,
-    Union,
 )
 from urllib.parse import unquote
 
@@ -26,9 +25,11 @@
 from fastapi_jsonapi.api import RoutersJSONAPI
 from fastapi_jsonapi.exceptions import (
     BadRequest,
+    InvalidField,
     InvalidFilters,
     InvalidInclude,
     InvalidSort,
+    InvalidType,
 )
 from fastapi_jsonapi.schema import (
     get_model_field,
@@ -89,30 +90,45 @@ def __init__(self, request: Request) -> None:
         self.MAX_INCLUDE_DEPTH: int = self.config.get("MAX_INCLUDE_DEPTH", 3)
         self.headers: HeadersQueryStringManager = HeadersQueryStringManager(**dict(self.request.headers))
 
-    def _get_key_values(self, name: str) -> Dict[str, Union[List[str], str]]:
+    def _extract_item_key(self, key: str) -> str:
+        try:
+            key_start = key.index("[") + 1
+            key_end = key.index("]")
+            return key[key_start:key_end]
+        except Exception:
+            msg = "Parse error"
+            raise BadRequest(msg, parameter=key)
+
+    def _get_unique_key_values(self, name: str) -> Dict[str, str]:
         """
         Return a dict containing key / values items for a given key, used for items like filters, page, etc.
 
         :param name: name of the querystring parameter
         :return: a dict of key / values items
         :raises BadRequest: if an error occurred while parsing the querystring.
         """
-        results = defaultdict(set)
+        results = {}
 
         for raw_key, value in self.qs.multi_items():
             key = unquote(raw_key)
-            try:
-                if not key.startswith(name):
-                    continue
+            if not key.startswith(name):
+                continue
 
-                key_start = key.index("[") + 1
-                key_end = key.index("]")
-                item_key = key[key_start:key_end]
+            item_key = self._extract_item_key(key)
+            results[item_key] = value
 
-                results[item_key].update(value.split(","))
-            except Exception:
-                msg = "Parse error"
-                raise BadRequest(msg, parameter=key)
+        return results
+
+    def _get_multiple_key_values(self, name: str) -> Dict[str, List]:
+        results = defaultdict(list)
+
+        for raw_key, value in self.qs.multi_items():
+            key = unquote(raw_key)
+            if not key.startswith(name):
+                continue
+
+            item_key = self._extract_item_key(key)
+            results[item_key].extend(value.split(","))
 
         return results
 
@@ -131,7 +147,7 @@ def querystring(self) -> Dict[str, str]:
         return {
             key: value
             for (key, value) in self.qs.multi_items()
-            if key.startswith(self.managed_keys) or self._get_key_values("filter[")
+            if key.startswith(self.managed_keys) or self._get_unique_key_values("filter[")
         }
 
     @property
@@ -156,8 +172,8 @@ def filters(self) -> List[dict]:
                 raise InvalidFilters(msg)
 
             results.extend(loaded_filters)
-        if self._get_key_values("filter["):
-            results.extend(self._simple_filters(self._get_key_values("filter[")))
+        if filter_key_values := self._get_unique_key_values("filter["):
+            results.extend(self._simple_filters(filter_key_values))
         return results
 
     @cached_property
@@ -183,7 +199,7 @@ def pagination(self) -> PaginationQueryStringManager:
         :raises BadRequest: if the client is not allowed to disable pagination.
         """
         # check values type
-        pagination_data: Dict[str, Union[List[str], str]] = self._get_key_values("page")
+        pagination_data: Dict[str, str] = self._get_unique_key_values("page")
         pagination = PaginationQueryStringManager(**pagination_data)
         if pagination_data.get("size") is None:
             pagination.size = None
@@ -213,23 +229,27 @@ def fields(self) -> Dict[str, List[str]]:
 
         :raises InvalidField: if result field not in schema.
         """
-        fields = self._get_key_values("fields")
+        fields = self._get_multiple_key_values("fields")
         for resource_type, field_names in fields.items():
             # TODO: we have registry for models (BaseModel)
             # TODO: create `type to schemas` registry
 
-            # schema: Type[BaseModel] = get_schema_from_type(key, self.app)
+            if resource_type not in RoutersJSONAPI.all_jsonapi_routers:
+                msg = f"Application has no resource with type {resource_type!r}"
+                raise InvalidType(msg)
+
+            schema: Type[BaseModel] = RoutersJSONAPI.all_jsonapi_routers[resource_type]._schema
             self._get_schema(resource_type)
 
-            # for field_name in field_names:
-            #     if field_name not in schema.__fields__:
-            #         msg = "{schema} has no attribute {field}".format(
-            #             schema=schema.__name__,
-            #             field=field_name,
-            #         )
-            #         raise InvalidField(msg)
+            for field_name in field_names:
+                if field_name not in schema.__fields__:
+                    msg = "{schema} has no attribute {field}".format(
+                        schema=schema.__name__,
+                        field=field_name,
+                    )
+                    raise InvalidField(msg)
 
-        return fields
+        return {resource_type: set(field_names) for resource_type, field_names in fields.items()}
 
     def _get_schema(self, resource_type: str) -> Type[BaseModel]:
         target_router = RoutersJSONAPI.all_jsonapi_routers[resource_type]

fastapi_jsonapi/views/list_view.py

@@ -1,12 +1,12 @@
 import logging
-from typing import TYPE_CHECKING, Any, Dict
+from typing import TYPE_CHECKING, Any, Dict, Union
 
 from fastapi_jsonapi.schema import (
     BaseJSONAPIItemInSchema,
     JSONAPIResultDetailSchema,
     JSONAPIResultListSchema,
 )
-from fastapi_jsonapi.views.utils import get_includes_indexes_by_type
+from fastapi_jsonapi.views.utils import handle_fields
 from fastapi_jsonapi.views.view_base import ViewBase
 
 if TYPE_CHECKING:
@@ -15,31 +15,6 @@
 logger = logging.getLogger(__name__)
 
 
-def calculate_include_fields(response, query_params, jsonapi) -> Dict:
-    included = "included" in response.__fields__ and response.included or []
-
-    include_params = {
-        field_name: {*response.__fields__[field_name].type_.__fields__.keys()}
-        for field_name in response.__fields__
-        if field_name
-    }
-    include_params["included"] = {}
-
-    includes_indexes_by_type = get_includes_indexes_by_type(included)
-
-    for resource_type, field_names in query_params.fields.items():
-        if resource_type == jsonapi.type_:
-            include_params["data"] = {"__all__": {"attributes": field_names, "id": {"id"}, "type": {"type"}}}
-            continue
-
-        target_type_indexes = includes_indexes_by_type.get(resource_type)
-
-        if resource_type in includes_indexes_by_type and target_type_indexes:
-            include_params["included"].update((idx, field_names) for idx in target_type_indexes)
-
-    return include_params
-
-
 class ListViewBase(ViewBase):
     def _calculate_total_pages(self, db_items_count: int) -> int:
         total_pages = 1
@@ -60,23 +35,14 @@ async def get_data_layer(
     ) -> "BaseDataLayer":
         return await self.get_data_layer_for_list(extra_view_deps)
 
-    async def handle_get_resource_list(self, **extra_view_deps) -> JSONAPIResultListSchema:
+    async def handle_get_resource_list(self, **extra_view_deps) -> Union[JSONAPIResultListSchema, Dict]:
         dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
         query_params = self.query_params
         count, items_from_db = await dl.get_collection(qs=query_params)
         total_pages = self._calculate_total_pages(count)
 
         response = self._build_list_response(items_from_db, count, total_pages)
-
-        if not query_params.fields:
-            return response
-
-        include_params = calculate_include_fields(response, query_params, self.jsonapi)
-
-        if include_params:
-            return response.dict(include=include_params)
-
-        return response
+        return handle_fields(response, query_params, self.jsonapi)
 
     async def handle_post_resource_list(
         self,

fastapi_jsonapi/views/utils.py

@@ -1,9 +1,39 @@
+from __future__ import annotations
+
 from collections import defaultdict
 from enum import Enum
 from functools import cache
-from typing import Callable, Coroutine, Dict, List, Optional, Set, Type, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Coroutine,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Set,
+    Type,
+    Union,
+)
 
 from pydantic import BaseModel
+from pydantic.fields import ModelField
+
+from fastapi_jsonapi.data_typing import TypeSchema
+from fastapi_jsonapi.schema import JSONAPIObjectSchema
+from fastapi_jsonapi.schema_builder import (
+    JSONAPIResultDetailSchema,
+    JSONAPIResultListSchema,
+)
+
+if TYPE_CHECKING:
+    from fastapi_jsonapi.api import RoutersJSONAPI
+    from fastapi_jsonapi.querystring import QueryStringManager
+
+
+JSONAPIResponse = Union[JSONAPIResultDetailSchema, JSONAPIResultListSchema]
+IGNORE_ALL_FIELDS_LITERAL = ""
 
 
 class HTTPMethod(Enum):
@@ -30,10 +60,97 @@ def handler(self) -> Optional[Union[Callable, Coroutine]]:
         return self.prepare_data_layer_kwargs
 
 
-def get_includes_indexes_by_type(included: List[Dict]) -> Dict[str, List[int]]:
+def _get_includes_indexes_by_type(included: List[JSONAPIObjectSchema]) -> Dict[str, List[int]]:
     result = defaultdict(list)
 
-    for idx, item in enumerate(included, 1):
-        result[item["type"]].append(idx)
+    for idx, item in enumerate(included):
+        result[item.type].append(idx)
 
     return result
+
+
+# TODO: move to schema builder?
+def _is_relationship_field(field: ModelField) -> bool:
+    return "relationship" in field.field_info.extra
+
+
+def _get_schema_field_names(schema: Type[TypeSchema]) -> Set[str]:
+    """
+    Returns all attribute names except relationships
+    """
+    result = set()
+
+    for field_name, field in schema.__fields__.items():
+        if _is_relationship_field(field):
+            continue
+
+        result.add(field_name)
+
+    return result
+
+
+def _get_exclude_fields(
+    schema: Type[TypeSchema],
+    include_fields: Iterable[str],
+) -> Set[str]:
+    schema_fields = _get_schema_field_names(schema)
+
+    if IGNORE_ALL_FIELDS_LITERAL in include_fields:
+        return schema_fields
+
+    return set(_get_schema_field_names(schema)).difference(include_fields)
+
+
+def _calculate_exclude_fields(
+    response: JSONAPIResponse,
+    query_params: QueryStringManager,
+    jsonapi: RoutersJSONAPI,
+) -> Dict:
+    included = "included" in response.__fields__ and response.included or []
+    is_list_response = isinstance(response, JSONAPIResultListSchema)
+
+    exclude_params: Dict[str, Any] = {}
+
+    includes_indexes_by_type = _get_includes_indexes_by_type(included)
+
+    for resource_type, field_names in query_params.fields.items():
+        schema = jsonapi.all_jsonapi_routers[resource_type]._schema
+        exclude_fields = _get_exclude_fields(schema, include_fields=field_names)
+        attributes_exclude = {"attributes": exclude_fields}
+
+        if resource_type == jsonapi.type_:
+            if is_list_response:
+                exclude_params["data"] = {"__all__": attributes_exclude}
+            else:
+                exclude_params["data"] = attributes_exclude
+
+            continue
+
+        if not included:
+            continue
+
+        target_type_indexes = includes_indexes_by_type.get(resource_type)
+
+        if target_type_indexes:
+            if "included" not in exclude_params:
+                exclude_params["included"] = {}
+
+            exclude_params["included"].update((idx, attributes_exclude) for idx in target_type_indexes)
+
+    return exclude_params
+
+
+def handle_fields(
+    response: JSONAPIResponse,
+    query_params: QueryStringManager,
+    jsonapi: RoutersJSONAPI,
+) -> Union[JSONAPIResponse, Dict]:
+    if not query_params.fields:
+        return response
+
+    exclude_params = _calculate_exclude_fields(response, query_params, jsonapi)
+
+    if exclude_params:
+        return response.dict(exclude=exclude_params)
+
+    return response