Add mypy to test step; fixes based on mypy errors

Author: lossyrob

mypy.ini

@@ -0,0 +1,3 @@
+[mypy]
+ignore_missing_imports = True
+disallow_untyped_defs = True

pystac/__init__.py

@@ -23,7 +23,7 @@ class STACTypeError(Exception):
 
 from typing import Any, Dict, Optional
 from pystac.version import (__version__, get_stac_version, set_stac_version)  # type:ignore
-from pystac.stac_io import STAC_IO
+from pystac.stac_io import STAC_IO  # type:ignore
 from pystac.extensions import Extensions  # type:ignore
 from pystac.stac_object import (STACObject, STACObjectType)  # type:ignore
 from pystac.media_type import MediaType  # type:ignore
@@ -37,12 +37,8 @@ class STACTypeError(Exception):
     Provider)  # type:ignore
 from pystac.item import (Item, Asset, CommonMetadata)  # type:ignore
 
-from pystac.serialization import stac_object_from_dict
-
 import pystac.validation
 
-STAC_IO.stac_object_from_dict = stac_object_from_dict
-
 import pystac.extensions.base
 import pystac.extensions.eo
 import pystac.extensions.label
@@ -92,7 +88,9 @@ def read_file(href: str) -> STACObject:
     return STACObject.from_file(href)
 
 
-def write_file(obj: STACObject, include_self_link: bool = True, dest_href: Optional[str] = None):
+def write_file(obj: STACObject,
+               include_self_link: bool = True,
+               dest_href: Optional[str] = None) -> None:
     """Writes a STACObject to a file.
 
     This will write only the Catalog, Collection or Item ``obj``. It will not attempt
@@ -115,7 +113,9 @@ def write_file(obj: STACObject, include_self_link: bool = True, dest_href: Optio
     obj.save_object(include_self_link=include_self_link, dest_href=dest_href)
 
 
-def read_dict(d: Dict[str, Any], href: Optional[str] = None, root: Optional[Catalog] = None):
+def read_dict(d: Dict[str, Any],
+              href: Optional[str] = None,
+              root: Optional[Catalog] = None) -> STACObject:
     """Reads a STAC object from a dict representing the serialized JSON version of the
     STAC object.
 
@@ -132,4 +132,4 @@ def read_dict(d: Dict[str, Any], href: Optional[str] = None, root: Optional[Cata
             If provided, the root's resolved object cache can be used to search for
             previously resolved instances of the STAC object.
     """
-    return stac_object_from_dict(d, href, root)
+    return STAC_IO.stac_object_from_dict(d, href, root)

pystac/cache.py

@@ -26,7 +26,7 @@ def get_cache_key(stac_object: "STACObject") -> Tuple[str, bool]:
         return (href, True)
     else:
         ids: List[str] = []
-        obj = stac_object
+        obj: Optional[STACObject] = stac_object
         while obj is not None:
             ids.append(obj.id)
             obj = obj.get_parent()
@@ -65,7 +65,7 @@ def __init__(self,
         self.hrefs_to_objects = hrefs_to_objects or {}
         self.ids_to_collections = ids_to_collections or {}
 
-        self._collection_cache = None
+        self._collection_cache: Optional[ResolvedObjectCollectionCache] = None
 
     def get_or_cache(self, obj: "STACObject") -> "STACObject":
         """Gets the STACObject that is the cached version of the given STACObject; or, if
@@ -268,7 +268,9 @@ def contains_id(self, collection_id: str) -> bool:
         return (self.resolved_object_cache.contains_collection_id(collection_id)
                 or super().contains_id(collection_id))
 
-    def cache(self, collection: Dict[str, Any], href: Optional[str] = None) -> None:
+    def cache(self,
+              collection: Union["Collection", Dict[str, Any]],
+              href: Optional[str] = None) -> None:
         super().cache(collection, href)
 
     @staticmethod

pystac/catalog.py

@@ -14,7 +14,7 @@
 
 
 class CatalogType(str, Enum):
-    def __str__(self):
+    def __str__(self) -> str:
         return str(self.value)
 
     SELF_CONTAINED = 'SELF_CONTAINED'
@@ -137,7 +137,7 @@ def __init__(self,
         if href is not None:
             self.set_self_href(href)
 
-        self.catalog_type = catalog_type
+        self.catalog_type: CatalogType = catalog_type
 
         self._resolved_objects.cache(self)
 
@@ -156,7 +156,7 @@ def is_relative(self) -> bool:
     def add_child(self,
                   child: "Catalog",
                   title: Optional[str] = None,
-                  strategy: Optional[HrefLayoutStrategy] = None):
+                  strategy: Optional[HrefLayoutStrategy] = None) -> None:
         """Adds a link to a child :class:`~pystac.Catalog` or :class:`~pystac.Collection`.
         This method will set the child's parent to this object, and its root to
         this Catalog's root.
@@ -268,15 +268,15 @@ def get_children(self) -> Iterable["Catalog"]:
         """
         return map(lambda x: cast(ps.Catalog, x), self.get_stac_objects('child'))
 
-    def get_child_links(self):
+    def get_child_links(self) -> List[Link]:
         """Return all child links of this catalog.
 
         Return:
             List[Link]: List of links of this catalog with ``rel == 'child'``
         """
         return self.get_links('child')
 
-    def clear_children(self) -> "Catalog":
+    def clear_children(self) -> None:
         """Removes all children from this catalog.
 
         Return:
@@ -285,7 +285,6 @@ def clear_children(self) -> "Catalog":
         child_ids = [child.id for child in self.get_children()]
         for child_id in child_ids:
             self.remove_child(child_id)
-        return self
 
     def remove_child(self, child_id: str) -> None:
         """Removes an child from this catalog.
@@ -336,7 +335,7 @@ def get_items(self) -> Iterable["ItemType"]:
         """
         return map(lambda x: cast(ps.Item, x), self.get_stac_objects('item'))
 
-    def clear_items(self) -> "Catalog":
+    def clear_items(self) -> None:
         """Removes all items from this catalog.
 
         Return:
@@ -349,7 +348,6 @@ def clear_items(self) -> "Catalog":
                 item.set_root(None)
 
         self.links = [link for link in self.links if link.rel != 'item']
-        return self
 
     def remove_item(self, item_id: str) -> None:
         """Removes an item from this catalog.
@@ -396,7 +394,7 @@ def get_item_links(self) -> List[Link]:
     def to_dict(self, include_self_link: bool = True) -> Dict[str, Any]:
         links = self.links
         if not include_self_link:
-            links = filter(lambda l: l.rel != 'self', links)
+            links = [x for x in links if x.rel != 'self']
 
         d: Dict[str, Any] = {
             'id': self.id,
@@ -438,15 +436,15 @@ def clone(self) -> "Catalog":
 
         return clone
 
-    def make_all_asset_hrefs_relative(self):
+    def make_all_asset_hrefs_relative(self) -> None:
         """Makes all the HREFs of assets belonging to items in this catalog
         and all children to be relative, recursively.
         """
         for _, _, items in self.walk():
             for item in items:
                 item.make_asset_hrefs_relative()
 
-    def make_all_asset_hrefs_absolute(self):
+    def make_all_asset_hrefs_absolute(self) -> None:
         """Makes all the HREFs of assets belonging to items in this catalog
         and all children to be absolute, recursively.
         """
@@ -457,7 +455,7 @@ def make_all_asset_hrefs_absolute(self):
     def normalize_and_save(self,
                            root_href: str,
                            catalog_type: Optional[CatalogType] = None,
-                           strategy: Optional[HrefLayoutStrategy] = None):
+                           strategy: Optional[HrefLayoutStrategy] = None) -> None:
         """Normalizes link HREFs to the given root_href, and saves the catalog.
 
         This is a convenience method that simply calls :func:`Catalog.normalize_hrefs
@@ -476,7 +474,9 @@ def normalize_and_save(self,
         self.normalize_hrefs(root_href, strategy=strategy)
         self.save(catalog_type)
 
-    def normalize_hrefs(self, root_href: str, strategy: Optional[HrefLayoutStrategy] = None):
+    def normalize_hrefs(self,
+                        root_href: str,
+                        strategy: Optional[HrefLayoutStrategy] = None) -> None:
         """Normalize HREFs will regenerate all link HREFs based on
         an absolute root_href and the canonical catalog layout as specified
         in the STAC specification's best practices.
@@ -492,7 +492,9 @@ def normalize_hrefs(self, root_href: str, strategy: Optional[HrefLayoutStrategy]
             `STAC best practices document <https://github.com/radiantearth/stac-spec/blob/v0.8.1/best-practices.md#catalog-layout>`_ for the canonical layout of a STAC.
         """ # noqa E501
         if strategy is None:
-            strategy = BestPracticesLayoutStrategy()
+            _strategy: HrefLayoutStrategy = BestPracticesLayoutStrategy()
+        else:
+            _strategy = strategy
 
         # Normalizing requires an absolute path
         if not is_absolute_href(root_href):
@@ -501,9 +503,9 @@ def normalize_hrefs(self, root_href: str, strategy: Optional[HrefLayoutStrategy]
         def process_item(item: "ItemType", _root_href: str) -> Callable[[], None]:
             item.resolve_links()
 
-            new_self_href = strategy.get_href(item, _root_href)
+            new_self_href = _strategy.get_href(item, _root_href)
 
-            def fn():
+            def fn() -> None:
                 item.set_self_href(new_self_href)
 
             return fn
@@ -514,7 +516,7 @@ def process_catalog(cat: Catalog, _root_href: str,
 
             cat.resolve_links()
 
-            new_self_href = strategy.get_href(cat, _root_href, is_root)
+            new_self_href = _strategy.get_href(cat, _root_href, is_root)
             new_root = os.path.dirname(new_self_href)
 
             for item in cat.get_items():
@@ -523,7 +525,7 @@ def process_catalog(cat: Catalog, _root_href: str,
             for child in cat.get_children():
                 setter_funcs.extend(process_catalog(child, new_root, is_root=False))
 
-            def fn():
+            def fn() -> None:
                 cat.set_self_href(new_self_href)
 
             setter_funcs.append(fn)
@@ -538,8 +540,6 @@ def fn():
         for fn in setter_funcs:
             fn()
 
-        return self
-
     def generate_subcatalogs(self,
                              template: str,
                              defaults: Optional[Dict[str, Any]] = None,
@@ -602,9 +602,9 @@ def generate_subcatalogs(self,
                 curr_parent = subcat
 
             # resolve collection link so when added back points to correct location
-            link = item.get_single_link('collection')
-            if link is not None:
-                link.resolve_stac_object()
+            col_link = item.get_single_link('collection')
+            if col_link is not None:
+                col_link.resolve_stac_object()
 
             curr_parent.add_item(item)
 
@@ -613,7 +613,7 @@ def generate_subcatalogs(self,
 
         return result
 
-    def save(self, catalog_type: CatalogType = None) -> None:
+    def save(self, catalog_type: Optional[CatalogType] = None) -> None:
         """Save this catalog and all it's children/item to files determined by the object's
         self link HREF.
 
@@ -652,14 +652,16 @@ def save(self, catalog_type: CatalogType = None) -> None:
 
         include_self_link = False
         # include a self link if this is the root catalog or if ABSOLUTE_PUBLISHED catalog
-        if ((self.get_self_href() == self.get_root_link().get_absolute_href()
-             and root.catalog_type != CatalogType.SELF_CONTAINED)
-                or root.catalog_type == CatalogType.ABSOLUTE_PUBLISHED):
+        if root.catalog_type == CatalogType.ABSOLUTE_PUBLISHED:
             include_self_link = True
+        elif root.catalog_type != CatalogType.SELF_CONTAINED:
+            root_link = self.get_root_link()
+            if root_link and root_link.get_absolute_href() == self.get_self_href():
+                include_self_link = True
 
         self.save_object(include_self_link=include_self_link)
-
-        self.catalog_type = catalog_type
+        if catalog_type is not None:
+            self.catalog_type = catalog_type
 
     def walk(self) -> Iterable[Tuple["Catalog", Iterable["Catalog"], Iterable["ItemType"]]]:
         """Walks through children and items of catalogs.
@@ -702,7 +704,9 @@ def validate_all(self) -> None:
     def _object_links(self) -> List[str]:
         return ['child', 'item'] + (ps.STAC_EXTENSIONS.get_extended_object_links(self))
 
-    def map_items(self, item_mapper: Callable[["ItemType"], Union["ItemType", List["ItemType"]]]):
+    def map_items(
+            self, item_mapper: Callable[["ItemType"], Union["ItemType",
+                                                            List["ItemType"]]]) -> "Catalog":
         """Creates a copy of a catalog, with each item passed through the
         item_mapper function.
 
@@ -718,7 +722,7 @@ def map_items(self, item_mapper: Callable[["ItemType"], Union["ItemType", List["
 
         new_cat = cast(Catalog, self.full_copy())
 
-        def process_catalog(catalog: Catalog):
+        def process_catalog(catalog: Catalog) -> None:
             for child in catalog.get_children():
                 process_catalog(child)
 
@@ -742,9 +746,10 @@ def process_catalog(catalog: Catalog):
         process_catalog(new_cat)
         return new_cat
 
-    def map_assets(self, asset_mapper: Callable[[str, "AssetType"],
-                                                Union["AssetType", Tuple[str, "AssetType"],
-                                                      Dict[str, "AssetType"]]]):
+    def map_assets(
+        self, asset_mapper: Callable[[str, "AssetType"], Union["AssetType", Tuple[str, "AssetType"],
+                                                               Dict[str, "AssetType"]]]
+    ) -> "Catalog":
         """Creates a copy of a catalog, with each Asset for each Item passed
         through the asset_mapper function.
 
@@ -758,7 +763,7 @@ def map_assets(self, asset_mapper: Callable[[str, "AssetType"],
             Catalog: A full copy of this catalog, with assets manipulated according
             to the asset_mapper function.
         """
-        def apply_asset_mapper(tup: Tuple[str, "AssetType"]):
+        def apply_asset_mapper(tup: Tuple[str, "AssetType"]) -> List[Tuple[str, ps.Asset]]:
             k, v = tup
             result = asset_mapper(k, v)
             if result is None:
@@ -773,7 +778,7 @@ def apply_asset_mapper(tup: Tuple[str, "AssetType"]):
                     raise Exception('asset_mapper must return a non-empty list')
                 return assets
 
-        def item_mapper(item: ps.Item):
+        def item_mapper(item: ps.Item) -> ps.Item:
             new_assets = [
                 x for result in map(apply_asset_mapper, item.assets.items()) for x in result
             ]
@@ -782,7 +787,7 @@ def item_mapper(item: ps.Item):
 
         return self.map_items(item_mapper)
 
-    def describe(self, include_hrefs: bool = False, _indent: int = 0):
+    def describe(self, include_hrefs: bool = False, _indent: int = 0) -> None:
         """Prints out information about this Catalog and all contained
         STACObjects.