Add some type annotations

asf_search/ASFProduct.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Dict, Tuple, Type, List, final
+from typing import Any, Dict, Optional, Tuple, Type, List, final
 import warnings
 from shapely.geometry import shape, Point, Polygon, mapping
 import json
@@ -158,8 +158,8 @@ def geojson(self) -> Dict:
     def download(
         self,
         path: str,
-        filename: str = None,
-        session: ASFSession = None,
+        filename: Optional[str] = None,
+        session: Optional[ASFSession] = None,
         fileType=FileDownloadType.DEFAULT_FILE,
     ) -> None:
         """
@@ -220,7 +220,7 @@ def get_urls(self, fileType=FileDownloadType.DEFAULT_FILE) -> list:
 
     def _get_additional_filenames_and_urls(
         self,
-        default_filename: str = None,  # for subclasses without fileName in url (see S1BurstProduct implementation) # noqa F401
+        default_filename: Optional[str] = None,  # for subclasses without fileName in url (see S1BurstProduct implementation) # noqa F401
     ) -> List[Tuple[str, str]]:
         return [
             (self._parse_filename_from_url(url), url)
@@ -233,7 +233,7 @@ def _parse_filename_from_url(self, url: str) -> str:
         return filename
 
     def stack(
-        self, opts: ASFSearchOptions = None, useSubclass: Type['ASFProduct'] = None
+        self, opts: Optional[ASFSearchOptions] = None, useSubclass: Optional[Type['ASFProduct']] = None
     ) -> ASFSearchResults:
         """
         Builds a baseline stack from this product.
@@ -257,7 +257,7 @@ def stack(
 
         return stack_from_product(self, opts=opts, ASFProductSubclass=useSubclass)
 
-    def get_stack_opts(self, opts: ASFSearchOptions = None) -> ASFSearchOptions:
+    def get_stack_opts(self, opts: Optional[ASFSearchOptions] = None) -> ASFSearchOptions:
         """
         Build search options that can be used to find an insar stack for this product
 
@@ -301,7 +301,7 @@ def _get_additional_urls(self) -> List[str]:
             and 's3credentials' not in url
         ]
 
-    def find_urls(self, extension: str = None, pattern: str = r'.*', directAccess: bool = False) -> List[str]:
+    def find_urls(self, extension: Optional[str] = None, pattern: str = r'.*', directAccess: bool = False) -> List[str]:
         """
         Searches for all urls matching a given extension and/or pattern
         param extension: the file extension to search for. (Defaults to `None`)
@@ -423,7 +423,7 @@ def _read_property(self, key: str, default: Any = None) -> Any:
 
     @final
     @staticmethod
-    def umm_get(item: Dict, *args):
+    def umm_get(item: Optional[Dict], *args):
         """
         Used to search for values in CMR UMM
 

asf_search/ASFSearchResults.py

@@ -1,12 +1,12 @@
 from collections import UserList
 from multiprocessing import Pool
 import json
-from typing import List
+from typing import List, Optional
 from asf_search import ASFSession, ASFSearchOptions
 from asf_search.download.file_download_type import FileDownloadType
 from asf_search.exceptions import ASFSearchError
 
-from asf_search import ASF_LOGGER
+from asf_search import ASF_LOGGER, ASFProduct
 from asf_search.export.csv import results_to_csv
 from asf_search.export.jsonlite import results_to_jsonlite
 from asf_search.export.jsonlite2 import results_to_jsonlite2
@@ -15,8 +15,8 @@
 from asf_search.export.metalink import results_to_metalink
 
 
-class ASFSearchResults(UserList):
-    def __init__(self, *args, opts: ASFSearchOptions = None):
+class ASFSearchResults(UserList[ASFProduct]):
+    def __init__(self, *args, opts: Optional[ASFSearchOptions] = None):
         super().__init__(*args)
         # Store it JUST so the user can access it (There might be zero products)
         # Each product will use their own reference to opts (but points to the same obj)
@@ -47,7 +47,7 @@ def jsonlite(self):
     def jsonlite2(self):
         return results_to_jsonlite2(self)
 
-    def find_urls(self, extension: str = None, pattern: str = r'.*', directAccess: bool = False) -> List[str]:
+    def find_urls(self, extension: Optional[str] = None, pattern: str = r'.*', directAccess: bool = False) -> List[str]:
         """Returns a flat list of all https or s3 urls from all results matching an extension and/or regex pattern
         param extension: the file extension to search for. (Defaults to `None`)
             - Example: '.tiff'
@@ -68,7 +68,7 @@ def __str__(self):
     def download(
         self,
         path: str,
-        session: ASFSession = None,
+        session: Optional[ASFSession] = None,
         processes: int = 1,
         fileType=FileDownloadType.DEFAULT_FILE,
     ) -> None:

asf_search/ASFSession.py

@@ -1,7 +1,8 @@
 from logging import warn
 import platform
-from typing import List, Union
+from typing import List, Optional, Union
 import requests
+import requests.cookies
 from requests.utils import get_netrc_auth
 from requests.auth import HTTPBasicAuth
 import http.cookiejar
@@ -14,13 +15,13 @@
 class ASFSession(requests.Session):
     def __init__(
         self,
-        edl_host: str = None,
-        edl_client_id: str = None,
-        asf_auth_host: str = None,
-        cmr_host: str = None,
-        cmr_collections: str = None,
-        auth_domains: List[str] = None,
-        auth_cookie_names: List[str] = None,
+        edl_host: Optional[str] = None,
+        edl_client_id: Optional[str] = None,
+        asf_auth_host: Optional[str] = None,
+        cmr_host: Optional[str] = None,
+        cmr_collections: Optional[str] = None,
+        auth_domains: Optional[List[str]] = None,
+        auth_cookie_names: Optional[List[str]] = None,
     ):
         """
         ASFSession is a subclass of `requests.Session`, and is meant to ease
@@ -301,7 +302,7 @@ def rebuild_auth(self, prepared_request: requests.Request, response: requests.Re
         if new_auth is not None:
             prepared_request.prepare_auth(new_auth)
 
-    def _get_domain(self, url: str):
+    def _get_domain(self, url: Optional[str]):
         return requests.utils.urlparse(url).hostname
 
     # multi-processing does an implicit copy of ASFSession objects,

asf_search/ASFStackableProduct.py

@@ -1,6 +1,6 @@
 from enum import Enum
 import copy
-from typing import Dict, Union
+from typing import Dict, Optional, Union
 from asf_search import ASFSession, ASFProduct
 from asf_search.ASFSearchOptions import ASFSearchOptions
 from asf_search.exceptions import ASFBaselineError
@@ -31,7 +31,7 @@ def __init__(self, args: Dict = {}, session: ASFSession = ASFSession()):
         super().__init__(args, session)
         self.baseline = self.get_baseline_calc_properties()
 
-    def get_baseline_calc_properties(self) -> Dict:
+    def get_baseline_calc_properties(self) -> Optional[Dict]:
         insarBaseline = self.umm_cast(
             float,
             self.umm_get(
@@ -48,8 +48,8 @@ def get_baseline_calc_properties(self) -> Dict:
 
         return {'insarBaseline': insarBaseline}
 
-    def get_stack_opts(self, opts: ASFSearchOptions = None):
-        stack_opts = ASFSearchOptions() if opts is None else copy(opts)
+    def get_stack_opts(self, opts: Optional[ASFSearchOptions] = None):
+        stack_opts = ASFSearchOptions() if opts is None else copy.copy(opts)
         stack_opts.processingLevel = self.get_default_baseline_product_type()
 
         if self.properties.get('insarStackId') in [None, 'NA', 0, '0']:

asf_search/search/baseline_search.py

@@ -1,4 +1,4 @@
-from typing import Type
+from typing import Optional, Type
 from asf_search.baseline.stack import get_baseline_from_stack
 from asf_search import ASF_LOGGER
 from copy import copy
@@ -22,8 +22,8 @@
 
 def stack_from_product(
     reference: ASFProduct,
-    opts: ASFSearchOptions = None,
-    ASFProductSubclass: Type[ASFProduct] = None,
+    opts: Optional[ASFSearchOptions] = None,
+    ASFProductSubclass: Optional[Type[ASFProduct]] = None,
 ) -> ASFSearchResults:
     """
     Finds a baseline stack from a reference ASFProduct
@@ -69,8 +69,8 @@ def stack_from_product(
 
 def stack_from_id(
     reference_id: str,
-    opts: ASFSearchOptions = None,
-    useSubclass: Type[ASFProduct] = None,
+    opts: Optional[ASFSearchOptions] = None,
+    useSubclass: Optional[Type[ASFProduct]] = None,
 ) -> ASFSearchResults:
     """
     Finds a baseline stack from a reference product ID

asf_search/search/geo_search.py

@@ -1,4 +1,4 @@
-from typing import Literal, Tuple, Union, Sequence
+from typing import Literal, Optional, Tuple, Union, Sequence
 import datetime
 from copy import copy
 
@@ -8,59 +8,59 @@
 
 
 def geo_search(
-    absoluteOrbit: Union[
+    absoluteOrbit: Optional[Union[
         int, Tuple[int, int], range, Sequence[Union[int, Tuple[int, int], range]]
-    ] = None,
-    asfFrame: Union[
+    ]] = None,
+    asfFrame: Optional[Union[
         int, Tuple[int, int], range, Sequence[Union[int, Tuple[int, int], range]]
-    ] = None,
-    beamMode: Union[str, Sequence[str]] = None,
-    beamSwath: Union[str, Sequence[str]] = None,
-    campaign: Union[str, Sequence[str]] = None,
-    maxDoppler: float = None,
-    minDoppler: float = None,
-    end: Union[datetime.datetime, str] = None,
-    maxFaradayRotation: float = None,
-    minFaradayRotation: float = None,
-    flightDirection: str = None,
-    flightLine: str = None,
-    frame: Union[int, Tuple[int, int], range, Sequence[Union[int, Tuple[int, int], range]]] = None,
-    granule_list: Union[str, Sequence[str]] = None,
-    groupID: Union[str, Sequence[str]] = None,
-    insarStackId: str = None,
-    instrument: Union[str, Sequence[str]] = None,
-    intersectsWith: str = None,
-    lookDirection: Union[str, Sequence[str]] = None,
-    offNadirAngle: Union[
+    ]] = None,
+    beamMode: Optional[Union[str, Sequence[str]]] = None,
+    beamSwath: Optional[Union[str, Sequence[str]]] = None,
+    campaign: Optional[Union[str, Sequence[str]]] = None,
+    maxDoppler: Optional[float] = None,
+    minDoppler: Optional[float] = None,
+    end: Optional[Union[datetime.datetime, str]] = None,
+    maxFaradayRotation: Optional[float] = None,
+    minFaradayRotation: Optional[float] = None,
+    flightDirection: Optional[str] = None,
+    flightLine: Optional[str] = None,
+    frame: Optional[Union[int, Tuple[int, int], range, Sequence[Union[int, Tuple[int, int], range]]]] = None,
+    granule_list: Optional[Union[str, Sequence[str]]] = None,
+    groupID: Optional[Union[str, Sequence[str]]] = None,
+    insarStackId: Optional[str] = None,
+    instrument: Optional[Union[str, Sequence[str]]] = None,
+    intersectsWith: Optional[str] = None,
+    lookDirection: Optional[Union[str, Sequence[str]]] = None,
+    offNadirAngle: Optional[Union[
         float, Tuple[float, float], Sequence[Union[float, Tuple[float, float]]]
-    ] = None,
-    platform: Union[str, Sequence[str]] = None,
-    polarization: Union[str, Sequence[str]] = None,
-    processingDate: Union[datetime.datetime, str] = None,
-    processingLevel: Union[str, Sequence[str]] = None,
-    product_list: Union[str, Sequence[str]] = None,
-    relativeOrbit: Union[
+    ]] = None,
+    platform: Optional[Union[str, Sequence[str]]] = None,
+    polarization: Optional[Union[str, Sequence[str]]] = None,
+    processingDate: Optional[Union[datetime.datetime, str]] = None,
+    processingLevel: Optional[Union[str, Sequence[str]]] = None,
+    product_list: Optional[Union[str, Sequence[str]]] = None,
+    relativeOrbit: Optional[Union[
         int, Tuple[int, int], range, Sequence[Union[int, Tuple[int, int], range]]
-    ] = None,
-    season: Tuple[int, int] = None,
-    start: Union[datetime.datetime, str] = None,
-    absoluteBurstID: Union[int, Sequence[int]] = None,
-    relativeBurstID: Union[int, Sequence[int]] = None,
-    fullBurstID: Union[str, Sequence[str]] = None,
-    temporalBaselineDays: Union[str, Sequence[str]] = None,
-    operaBurstID: Union[str, Sequence[str]] = None,
-    frameCoverage: Literal["FULL", "PARTIAL"] = None,
-    mainBandPolarization: Union[str, Sequence[str]] = None,
-    sideBandPolarization: Union[str, Sequence[str]] = None,
-    rangeBandwidth: Union[str, Sequence[str]] = None,
-    jointObservation: bool = None,
-    productionConfiguration: Union[Literal["PR", "UR"], Sequence[Literal["PR", "UR"]]] = None,
-    dataset: Union[str, Sequence[str]] = None,
-    collections: Union[str, Sequence[str]] = None,
-    shortName: Union[str, Sequence[str]] = None,
-    cmr_keywords: Union[Tuple[str, str], Sequence[Tuple[str, str]]] = None,
-    maxResults: int = None,
-    opts: ASFSearchOptions = None,
+    ]] = None,
+    season: Optional[Tuple[int, int]] = None,
+    start: Optional[Union[datetime.datetime, str]] = None,
+    absoluteBurstID: Optional[Union[int, Sequence[int]]] = None,
+    relativeBurstID: Optional[Union[int, Sequence[int]]] = None,
+    fullBurstID: Optional[Union[str, Sequence[str]]] = None,
+    temporalBaselineDays: Optional[Union[str, Sequence[str]]] = None,
+    operaBurstID: Optional[Union[str, Sequence[str]]] = None,
+    frameCoverage: Optional[Literal["FULL", "PARTIAL"]] = None,
+    mainBandPolarization: Optional[Union[str, Sequence[str]]] = None,
+    sideBandPolarization: Optional[Union[str, Sequence[str]]] = None,
+    rangeBandwidth: Optional[Union[str, Sequence[str]]] = None,
+    jointObservation: Optional[bool] = None,
+    productionConfiguration: Optional[Union[Literal["PR", "UR"], Sequence[Literal["PR", "UR"]]]] = None,
+    dataset: Optional[Union[str, Sequence[str]]] = None,
+    collections: Optional[Union[str, Sequence[str]]] = None,
+    shortName: Optional[Union[str, Sequence[str]]] = None,
+    cmr_keywords: Optional[Union[Tuple[str, str], Sequence[Tuple[str, str]]]] = None,
+    maxResults: Optional[int] = None,
+    opts: Optional[ASFSearchOptions] = None,
 ) -> ASFSearchResults:
     """
     Performs a geographic search against the Central Metadata Repository (CMR),