improve typing

Author: vincentsarago

stac_fastapi/api/stac_fastapi/api/middleware.py

@@ -82,7 +82,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
         await self.app(scope, receive, send)
 
-    def _get_forwarded_url_parts(self, scope: Scope) -> Tuple[str]:
+    def _get_forwarded_url_parts(self, scope: Scope) -> Tuple[str, str, str]:
         proto = scope.get("scheme", "http")
         header_host = self._get_header_value_by_name(scope, "host")
         if header_host is None:
@@ -127,7 +127,7 @@ def _get_header_value_by_name(
     @staticmethod
     def _replace_header_value_by_name(
         scope: Scope, header_name: str, new_value: str
-    ) -> List[Tuple[str]]:
+    ) -> List[Tuple[str, str]]:
         return [
             (name, value)
             for name, value in scope["headers"]

stac_fastapi/extensions/stac_fastapi/extensions/core/aggregation/types.py

@@ -2,33 +2,34 @@
 
 from typing import Any, Dict, List, Literal, Optional, Union
 
-from pydantic import Field
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
 from stac_fastapi.types.rfc3339 import DateTimeType
 
+Bucket = TypedDict(
+    "Bucket",
+    {
+        "key": str,
+        "data_type": str,
+        "frequency": NotRequired[Dict],
+        # we can't use the `class Bucket` notation because `from` is a reserved key
+        "from": NotRequired[Union[int, float]],
+        "to": NotRequired[Optional[Union[int, float]]],
+    },
+)
 
-class Bucket(TypedDict, total=False):
-    """A STAC aggregation bucket."""
 
-    key: str
-    data_type: str
-    frequency: Optional[Dict] = None
-    _from: Optional[Union[int, float]] = Field(alias="from", default=None)
-    to: Optional[Optional[Union[int, float]]] = None
-
-
-class Aggregation(TypedDict, total=False):
+class Aggregation(TypedDict):
     """A STAC aggregation."""
 
     name: str
     data_type: str
-    buckets: Optional[List[Bucket]] = None
-    overflow: Optional[int] = None
-    value: Optional[Union[str, int, DateTimeType]] = None
+    buckets: NotRequired[List[Bucket]]
+    overflow: NotRequired[int]
+    value: NotRequired[Union[str, int, DateTimeType]]
 
 
-class AggregationCollection(TypedDict, total=False):
+class AggregationCollection(TypedDict):
     """STAC Item Aggregation Collection."""
 
     type: Literal["AggregationCollection"]

stac_fastapi/extensions/stac_fastapi/extensions/core/collection_search/request.py

@@ -37,15 +37,15 @@ class BaseCollectionSearchGetRequest(APIRequest, DatetimeMixin):
 class BaseCollectionSearchPostRequest(BaseModel):
     """Collection-Search POST model."""
 
-    bbox: Optional[BBox] = Field(
+    bbox: Optional[BBox] = Field(  # type: ignore
         default=None,
         description="Only return items intersecting this bounding box. Mutually exclusive with **intersects**.",  # noqa: E501
         openapi_examples={
             "user-provided": {"value": None},
             "Montreal": {"value": "-73.896103,45.364690,-73.413734,45.674283"},
         },
     )
-    datetime: Optional[str] = Field(
+    datetime: Optional[str] = Field(  # type: ignore
         default=None,
         description="""Only return items that have a temporal property that intersects this value.\n
 Either a date-time or an interval, open or closed. Date and time expressions adhere to RFC 3339. Open intervals are expressed using double-dots.""",  # noqa: E501

stac_fastapi/extensions/stac_fastapi/extensions/core/filter/request.py

@@ -50,7 +50,7 @@ class FilterExtensionGetRequest(APIRequest):
 class FilterExtensionPostRequest(BaseModel):
     """Filter extension POST request model."""
 
-    filter_expr: Optional[Dict[str, Any]] = Field(
+    filter_expr: Optional[Dict[str, Any]] = Field(  # type: ignore
         default=None,
         alias="filter",
         description="A CQL filter expression for filtering items.",

stac_fastapi/extensions/stac_fastapi/extensions/core/free_text/request.py

@@ -7,7 +7,7 @@
 from pydantic import BaseModel, Field
 from typing_extensions import Annotated
 
-from stac_fastapi.types.search import APIRequest, str2list
+from stac_fastapi.types.search import APIRequest
 
 
 def _ft_converter(
@@ -22,7 +22,9 @@ def _ft_converter(
         ),
     ] = None,
 ) -> Optional[List[str]]:
-    return str2list(val)
+    if val:
+        return val.split(",")
+    return None
 
 
 @attr.s

stac_fastapi/extensions/stac_fastapi/extensions/core/query/request.py

@@ -29,7 +29,7 @@ class QueryExtensionGetRequest(APIRequest):
 class QueryExtensionPostRequest(BaseModel):
     """Query Extension POST request model."""
 
-    query: Optional[Dict[str, Dict[str, Any]]] = Field(
+    query: Optional[Dict[str, Dict[str, Any]]] = Field(  # type: ignore
         None,
         description="Allows additional filtering based on the properties of Item objects",  # noqa: E501
         openapi_examples={

stac_fastapi/extensions/stac_fastapi/extensions/core/sort/request.py

@@ -37,7 +37,7 @@ class SortExtensionGetRequest(APIRequest):
 class SortExtensionPostRequest(BaseModel):
     """Sortby parameter for POST requests."""
 
-    sortby: Optional[List[PostSortModel]] = Field(
+    sortby: Optional[List[PostSortModel]] = Field(  # type: ignore
         None,
         description="An array of property (field) names, and direction in form of '{'field': '<property_name>', 'direction':'<direction>'}'",  # noqa: E501
         openapi_examples={

stac_fastapi/types/stac_fastapi/types/core.py

@@ -533,7 +533,7 @@ def item_collection(
         bbox: Optional[BBox] = None,
         datetime: Optional[str] = None,
         limit: int = 10,
-        token: str = None,
+        token: Optional[str] = None,
         **kwargs,
     ) -> stac.ItemCollection:
         """Get all items from a specific collection.
@@ -744,7 +744,7 @@ async def item_collection(
         bbox: Optional[BBox] = None,
         datetime: Optional[str] = None,
         limit: int = 10,
-        token: str = None,
+        token: Optional[str] = None,
         **kwargs,
     ) -> stac.ItemCollection:
         """Get all items from a specific collection.

stac_fastapi/types/stac_fastapi/types/extension.py

@@ -15,7 +15,7 @@ class ApiExtension(abc.ABC):
     GET = None
     POST = None
 
-    def get_request_model(self, verb: Optional[str] = "GET") -> Optional[BaseModel]:
+    def get_request_model(self, verb: str = "GET") -> Optional[BaseModel]:
         """Return the request model for the extension.method.
 
         The model can differ based on HTTP verb

stac_fastapi/types/stac_fastapi/types/rfc3339.py

@@ -145,7 +145,7 @@ def str_to_interval(interval: Optional[str]) -> Optional[DateTimeType]:
             status_code=400, detail="Start datetime cannot be before end datetime."
         )
 
-    return start, end
+    return start, end  # type: ignore
 
 
 def now_in_utc() -> datetime: