Typecheck some API interfaces

.github/workflows/typecheck.yml

@@ -0,0 +1,29 @@
+name: Typecheck
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+      - development
+
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Install package and test dependencies
+        run: |
+          python -m pip install .[complete]
+          python -m pip install -r requirements-dev.txt
+
+      - uses: jakebailey/pyright-action@v2

icepyx/core/APIformatting.py

@@ -1,6 +1,13 @@
-# Generate and format information for submitting to API (CMR and NSIDC)
+"""Generate and format information for submitting to API (CMR and NSIDC)."""
 
 import datetime as dt
+from typing import Any, Generic, Literal, TypeVar, Union, overload
+
+from icepyx.core.types import (
+    CMRParams,
+    EGIParamsSubset,
+    EGISpecificRequiredParams,
+)
 
 # ----------------------------------------------------------------------
 # parameter-specific formatting for display
@@ -183,12 +190,56 @@ def to_string(params):
     return "&".join(param_list)
 
 
+ParameterType = Literal["CMR", "required", "subset"]
+# DevGoal: When Python 3.12 is minimum supported version, migrate to PEP695 style
+T = TypeVar("T", bound=ParameterType)
+
+
+class _FmtedKeysDescriptor:
+    """Enable the Parameters class' fmted_keys property to be typechecked correctly.
+
+    See: https://github.com/microsoft/pyright/issues/3071#issuecomment-1043978070
+    """
+
+    @overload
+    def __get__(
+        self,
+        instance: 'Parameters[Literal["CMR"]]',
+        owner: Any,
+    ) -> CMRParams: ...
+
+    @overload
+    def __get__(
+        self,
+        instance: 'Parameters[Literal["required"]]',
+        owner: Any,
+    ) -> EGISpecificRequiredParams: ...
+
+    @overload
+    def __get__(
+        self,
+        instance: 'Parameters[Literal["subset"]]',
+        owner: Any,
+    ) -> EGIParamsSubset: ...
+
+    def __get__(
+        self,
+        instance: "Parameters",
+        owner: Any,
+    ) -> Union[CMRParams, EGISpecificRequiredParams, EGIParamsSubset]:
+        """
+        Returns the dictionary of formatted keys associated with the
+        parameter object.
+        """
+        return instance._fmted_keys
+
+
 # ----------------------------------------------------------------------
 # DevNote: Currently, this class is not tested!!
 # DevGoal: this could be expanded, similar to the variables class, to provide users with valid options if need be
 # DevGoal: currently this does not do much by way of checking/formatting of other subsetting options (reprojection or formats)
 # it would be great to incorporate that so that people can't just feed any keywords in...
-class Parameters:
+class Parameters(Generic[T]):
     """
     Build and update the parameter lists needed to submit a data order
 
@@ -206,7 +257,14 @@ class Parameters:
         on the type of query. Must be one of ['search','download']
     """
 
-    def __init__(self, partype, values=None, reqtype=None):
+    fmted_keys = _FmtedKeysDescriptor()
+
+    def __init__(
+        self,
+        partype: T,
+        values=None,
+        reqtype=None,
+    ):
         assert partype in [
             "CMR",
             "required",
@@ -242,15 +300,7 @@ def poss_keys(self):
 
     #     return self._wanted
 
-    @property
-    def fmted_keys(self):
-        """
-        Returns the dictionary of formatted keys associated with the
-        parameter object.
-        """
-        return self._fmted_keys
-
-    def _get_possible_keys(self):
+    def _get_possible_keys(self) -> dict[str, list[str]]:
         """
         Use the parameter type to get a list of possible parameter keys.
         """
@@ -347,7 +397,7 @@ def check_values(self):
         else:
             return False
 
-    def build_params(self, **kwargs):
+    def build_params(self, **kwargs) -> None:
         """
         Build the parameter dictionary of formatted key:value pairs for submission to NSIDC
         in the data request.
@@ -443,3 +493,8 @@ def build_params(self, **kwargs):
                             k = "Boundingshape"
 
                     self._fmted_keys.update({k: kwargs["spatial_extent"]})
+
+
+CMRParameters = Parameters[Literal["CMR"]]
+RequiredParameters = Parameters[Literal["required"]]
+SubsetParameters = Parameters[Literal["subset"]]

icepyx/core/auth.py

@@ -54,7 +54,7 @@ def __init__(self, auth=None):
         self._s3login_credentials = None
         self._s3_initial_ts = None  # timer for 1h expiration on s3 credentials
 
-    def __str__(self):
+    def __str__(self) -> str:
         if self.session:
             repr_string = "EarthdataAuth obj with session initialized"
         else:

icepyx/core/granules.py

@@ -1,19 +1,24 @@
+from __future__ import annotations
+
 import datetime
 import io
 import json
 import os
 import pprint
 import re
 import time
+from typing import Optional
 from xml.etree import ElementTree as ET
 import zipfile
 
 import numpy as np
 import requests
+from requests.compat import unquote
 
 import icepyx.core.APIformatting as apifmt
 from icepyx.core.auth import EarthdataAuthMixin
 import icepyx.core.exceptions
+from icepyx.core.types import CMRParams, EGISpecificRequiredParams
 from icepyx.core.urls import DOWNLOAD_BASE_URL, GRANULE_SEARCH_BASE_URL, ORDER_BASE_URL
 
 
@@ -170,14 +175,19 @@
     # ----------------------------------------------------------------------
     # Methods
 
-    def get_avail(self, CMRparams, reqparams, cloud=False):
+    def get_avail(
+        self,
+        CMRparams: Optional[CMRParams],
+        reqparams: Optional[EGISpecificRequiredParams],
+        cloud=False,
+    ):
         """
         Get a list of available granules for the query object's parameters.
         Generates the `avail` attribute of the granules object.
 
         Parameters
         ----------
-        CMRparams : dictionary
+        CMRparams :
             Dictionary of properly formatted CMR search parameters.
         reqparams : dictionary
             Dictionary of properly formatted parameters required for searching, ordering,
@@ -261,13 +271,13 @@
     # DevGoal: add kwargs to allow subsetting and more control over request options.
     def place_order(
         self,
-        CMRparams,
-        reqparams,
+        CMRparams: CMRParams,
+        reqparams: EGISpecificRequiredParams,
         subsetparams,
         verbose,
         subset=True,
         geom_filepath=None,
-    ):  # , **kwargs):
+    ):
         """
         Place an order for the available granules for the query object.
         Adds the list of zipped files (orders) to the granules data object (which is
@@ -276,11 +286,11 @@
 
         Parameters
         ----------
-        CMRparams : dictionary
+        CMRparams :
             Dictionary of properly formatted CMR search parameters.
-        reqparams : dictionary
+        reqparams :
             Dictionary of properly formatted parameters required for searching, ordering,
-            or downloading from NSIDC.
+            or downloading from NSIDC (via their EGI system).
         subsetparams : dictionary
             Dictionary of properly formatted subsetting parameters. An empty dictionary
             is passed as input here when subsetting is set to False in query methods.
@@ -359,7 +369,7 @@
             request.raise_for_status()
             esir_root = ET.fromstring(request.content)
             if verbose is True:
-                print("Order request URL: ", requests.utils.unquote(request.url))
+                print("Order request URL: ", unquote(request.url))
                 print(
                     "Order request response XML content: ",
                     request.content.decode("utf-8"),
@@ -402,6 +412,7 @@
                 loop_root = ET.fromstring(loop_response.content)
 
             # Continue loop while request is still processing
+            loop_root = None
             while status == "pending" or status == "processing":
                 print(
                     "Your order status is still ",
@@ -425,6 +436,13 @@
                 if status == "pending" or status == "processing":
                     continue
 
+            if not isinstance(loop_root, ET.Element):
+                # The typechecker determined that loop_root could be unbound at this
+                # point. We know for sure this shouldn't be possible, though, because
+                # the while loop should run once.
+                # See: https://github.com/microsoft/pyright/discussions/2033
+                raise RuntimeError("Programmer error!")
+
             # Order can either complete, complete_with_errors, or fail:
             # Provide complete_with_errors error message:
             if status == "complete_with_errors" or status == "failed":

icepyx/core/query.py

@@ -1,7 +1,9 @@
 import pprint
+from typing import Optional, Union
 
 import geopandas as gpd
 import matplotlib.pyplot as plt
+from typing_extensions import Never
 
 import icepyx.core.APIformatting as apifmt
 from icepyx.core.auth import EarthdataAuthMixin
@@ -11,6 +13,7 @@
 import icepyx.core.is2ref as is2ref
 import icepyx.core.spatial as spat
 import icepyx.core.temporal as tp
+from icepyx.core.types import CMRParams, EGIParamsSubset, EGISpecificRequiredParams
 import icepyx.core.validate_inputs as val
 from icepyx.core.variables import Variables as Variables
 from icepyx.core.visualization import Visualize
@@ -393,6 +396,10 @@
     GenQuery
     """
 
+    _CMRparams: apifmt.CMRParameters
+    _reqparams: apifmt.RequiredParameters
+    _subsetparams: Optional[apifmt.SubsetParameters]
+
     # ----------------------------------------------------------------------
     # Constructors
 
@@ -532,7 +539,7 @@
             return sorted(set(self._tracks))
 
     @property
-    def CMRparams(self):
+    def CMRparams(self) -> CMRParams:
         """
         Display the CMR key:value pairs that will be submitted.
         It generates the dictionary if it does not already exist.
@@ -573,7 +580,7 @@
         return self._CMRparams.fmted_keys
 
     @property
-    def reqparams(self):
+    def reqparams(self) -> EGISpecificRequiredParams:
         """
         Display the required key:value pairs that will be submitted.
         It generates the dictionary if it does not already exist.
@@ -599,7 +606,7 @@
     # @property
     # DevQuestion: if I make this a property, I get a "dict" object is not callable
     # when I try to give input kwargs... what approach should I be taking?
-    def subsetparams(self, **kwargs):
+    def subsetparams(self, **kwargs) -> Union[EGIParamsSubset, dict[Never, Never]]:
         """
         Display the subsetting key:value pairs that will be submitted.
         It generates the dictionary if it does not already exist
@@ -1001,7 +1008,7 @@
         if "email" in self._reqparams.fmted_keys or email is False:
             self._reqparams.build_params(**self._reqparams.fmted_keys)
         elif email is True:
-            user_profile = self.auth.get_user_profile()
+            user_profile = self.auth.get_user_profile()  # pyright: ignore[reportAttributeAccessIssue]
             self._reqparams.build_params(
                 **self._reqparams.fmted_keys, email=user_profile["email_address"]
             )
@@ -1135,14 +1142,14 @@
             import geoviews as gv
             from shapely.geometry import Polygon  # noqa: F401
 
-            gv.extension("bokeh")
+            gv.extension("bokeh")  # pyright: ignore[reportCallIssue]
 
             bbox_poly = gv.Path(gdf["geometry"]).opts(color="red", line_color="red")
             tile = gv.tile_sources.EsriImagery.opts(width=500, height=500)
-            return tile * bbox_poly
+            return tile * bbox_poly  # pyright: ignore[reportOperatorIssue]
 
         except ImportError:
-            world = gpd.read_file(gpd.datasets.get_path("naturalearth_lowres"))
+            world = gpd.read_file(gpd.datasets.get_path("naturalearth_lowres"))  # pyright: ignore[reportAttributeAccessIssue]
             f, ax = plt.subplots(1, figsize=(12, 6))
             world.plot(ax=ax, facecolor="lightgray", edgecolor="gray")
             gdf.plot(ax=ax, color="#FF8C00", alpha=0.7)