feat(#184): simplified JSON encoding

src/stac_fastapi/indexed/search/query_info.py

@@ -1,6 +1,5 @@
 from dataclasses import asdict, dataclass, replace
-from json import JSONEncoder, loads
-from typing import Any, Dict, Final, List, Optional, Self, Type, cast
+from typing import Any, Dict, Final, List, Optional, cast
 
 from geojson_pydantic.geometries import parse_geometry_obj
 from stac_pydantic.api.extensions.sort import SortExtension
@@ -47,28 +46,37 @@ def previous(self) -> "QueryInfo":
         )
 
     def to_dict(self) -> Dict[str, Any]:
-        return asdict(self)
-
-    @staticmethod
-    def json_encoder() -> Type:
-        return _CustomJSONEncoder
-
-    def json_post_decoder(self: Self) -> "QueryInfo":
-        return replace(
-            self,
-            intersects=parse_geometry_obj(loads(cast(str, self.intersects)))
+        return {
+            **asdict(self),
+            "intersects": cast(Intersection, self.intersects).model_dump()
             if self.intersects is not None
             else None,
-            order=[SortExtension(**loads(cast(str, entry))) for entry in self.order]
+            "order": [
+                {
+                    **cast(SortExtension, entry).model_dump(),
+                    **{"direction": cast(SortExtension, entry).direction.value},
+                }
+                for entry in self.order
+            ]
             if self.order is not None
             else None,
-        )
-
+        }
 
-class _CustomJSONEncoder(JSONEncoder):
-    def default(self, obj: Any) -> Any:
-        if isinstance(obj, Intersection):
-            return cast(Intersection, obj).model_dump_json()
-        if isinstance(obj, SortExtension):
-            return cast(SortExtension, obj).model_dump_json()
-        return JSONEncoder.default(self, obj)
+    @classmethod
+    def from_dict(cls, source_dict: Dict[str, Any]) -> "QueryInfo":
+        intersects_value = source_dict.get("intersects")
+        order_value = source_dict.get("order")
+        return cls(
+            **{
+                **source_dict,
+                "intersects": parse_geometry_obj(intersects_value)
+                if intersects_value is not None
+                else None,
+                "order": [
+                    SortExtension(**cast(Dict[str, Any], entry))
+                    for entry in order_value
+                ]
+                if order_value is not None
+                else None,
+            }
+        )

src/stac_fastapi/indexed/search/token.py

@@ -16,13 +16,13 @@ def get_query_info_from_token(
     token: str,
 ) -> QueryInfo:
     try:
-        result = QueryInfo(
-            **decode(
+        result = QueryInfo.from_dict(
+            decode(
                 jwt=token,
                 key=get_settings().token_jwt_secret,
                 algorithms=[_hashing_algorithm],
             )
-        ).json_post_decoder()
+        )
         if result.query_version != current_query_version:
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
@@ -39,5 +39,4 @@ def create_token_from_query(query_info: QueryInfo) -> str:
         payload=(query_info.to_dict()),
         key=get_settings().token_jwt_secret,
         algorithm=_hashing_algorithm,
-        json_encoder=QueryInfo.json_encoder(),
     )