Add basic ItemCollection implementation

CHANGELOG.md

@@ -8,6 +8,8 @@
 - Links to Issues, Discussions, and documentation sites ([#409](https://github.com/stac-utils/pystac/pull/409))
 - Python minimum version set to `>=3.6` ([#409](https://github.com/stac-utils/pystac/pull/409))
 - Code of Conduct ([#399](https://github.com/stac-utils/pystac/pull/399))
+- `ItemCollection` class for working with GeoJSON FeatureCollections containing only
+  STAC Items ([#430](https://github.com/stac-utils/pystac/pull/430))
 
 ### Changed
 

docs/api.rst

@@ -127,6 +127,14 @@ CommonMetadata
    :members:
    :undoc-members:
 
+ItemCollection
+--------------
+Represents a GeoJSON FeatureCollection in which all Features are STAC Items
+
+.. autoclass:: pystac.ItemCollection
+   :members:
+   :show-inheritance:
+
 Links
 -----
 

docs/conf.py

@@ -52,6 +52,7 @@
 extensions = [
     'sphinx.ext.autodoc',
     'sphinx.ext.viewcode',
+    'sphinx.ext.intersphinx',
     'sphinx.ext.napoleon',
     'sphinx.ext.githubpages',
     'sphinx.ext.extlinks',
@@ -207,3 +208,7 @@
 
 
 # -- Extension configuration -------------------------------------------------
+
+intersphinx_mapping = {
+    'python': ('https://docs.python.org/3', None),
+}

pystac/__init__.py

@@ -34,6 +34,7 @@
 )
 from pystac.summaries import RangeSummary
 from pystac.item import Item, Asset, CommonMetadata
+from pystac.item_collection import ItemCollection
 
 import pystac.validation
 
@@ -85,12 +86,20 @@ def read_file(href: str) -> STACObject:
     Returns:
         The specific STACObject implementation class that is represented
         by the JSON read from the file located at HREF.
+
+    Raises:
+        STACTypeError : If the file at ``href`` does not represent a valid
+            :class:`~pystac.STACObject`. Note that an :class:`~pystac.ItemCollection` is not
+            a :class:`~pystac.STACObject` and must be read using
+            :meth:`ItemCollection.from_file <pystac.ItemCollection.from_file>`
     """
     return STACObject.from_file(href)
 
 
 def write_file(
-    obj: STACObject, include_self_link: bool = True, dest_href: Optional[str] = None
+    obj: STACObject,
+    include_self_link: bool = True,
+    dest_href: Optional[str] = None,
 ) -> None:
     """Writes a STACObject to a file.
 
@@ -106,10 +115,10 @@ def write_file(
 
     Args:
         obj : The STACObject to save.
-        include_self_link : If this is true, include the 'self' link with this object.
+        include_self_link : If ``True``, include the ``"self"`` link with this object.
             Otherwise, leave out the self link.
-        dest_href : Optional HREF to save the file to. If None, the object will be saved
-            to the object's self href.
+        dest_href : Optional HREF to save the file to. If ``None``, the object will be
+            saved to the object's ``"self"`` href.
     """
     obj.save_object(include_self_link=include_self_link, dest_href=dest_href)
 
@@ -120,13 +129,14 @@ def read_dict(
     root: Optional[Catalog] = None,
     stac_io: Optional[StacIO] = None,
 ) -> STACObject:
-    """Reads a STAC object from a dict representing the serialized JSON version of the
-    STAC object.
+    """Reads a :class:`~STACObject` or :class:`~ItemCollection` from a JSON-like dict
+    representing a serialized STAC object.
 
-    This method will return either a Catalog, a Collection, or an Item based on what the
-    dict contains.
+    This method will return either a :class:`~Catalog`, :class:`~Collection`,
+    or :class`~Item` based on the contents of the dict.
 
-    This is a convenience method for :meth:`pystac.serialization.stac_object_from_dict`
+    This is a convenience method for either
+    :meth:`stac_io.stac_object_from_dict <stac_io.stac_object_from_dict>`.
 
     Args:
         d : The dict to parse.
@@ -135,8 +145,14 @@ def read_dict(
         root : Optional root of the catalog for this object.
             If provided, the root's resolved object cache can be used to search for
             previously resolved instances of the STAC object.
-        stac_io: Optional StacIO instance to use for reading. If None, the
-            default instance will be used.
+        stac_io: Optional :class:`~StacIO` instance to use for reading. If ``None``,
+            the default instance will be used.
+
+    Raises:
+        STACTypeError : If the ``d`` dictionary does not represent a valid
+            :class:`~pystac.STACObject`. Note that an :class:`~pystac.ItemCollection` is not
+            a :class:`~pystac.STACObject` and must be read using
+            :meth:`ItemCollection.from_dict <pystac.ItemCollection.from_dict>`
     """
     if stac_io is None:
         stac_io = StacIO.default()

pystac/item_collection.py

@@ -0,0 +1,212 @@
+from copy import deepcopy
+from pystac.errors import STACTypeError
+from typing import Any, Dict, Iterator, List, Optional, Collection, Iterable, Union
+
+import pystac
+from pystac.utils import make_absolute_href, is_absolute_href
+from pystac.serialization.identify import identify_stac_object_type
+
+
+ItemLike = Union[pystac.Item, Dict[str, Any]]
+
+
+class ItemCollection(Collection[pystac.Item]):
+    """Implementation of a GeoJSON FeatureCollection whose features are all STAC
+    Items.
+
+    All :class:`~pystac.Item` instances passed to the :class:`~ItemCollection` instance
+    during instantiation are cloned and have their ``"root"`` URL cleared. Instances of
+    this class implement the abstract methods of :class:`typing.Collection` and can also
+    be added together (see below for examples using these methods).
+
+    Any additional top-level fields in the FeatureCollection are retained in
+    :attr:`~ItemCollection.extra_fields` by the :meth:`~ItemCollection.from_dict` and
+    :meth:`~ItemCollection.from_file` methods and will be present in the serialized file
+    from :meth:`~ItemCollection.save_object`.
+
+    Arguments:
+
+        items : List of :class:`~pystac.Item` instances to include in the
+            :class:`~ItemCollection`.
+        extra_fields : Dictionary of additional top-level fields included in the
+            :class:`~ItemCollection`.
+        clone_items : Optional flag indicating whether :class:`~pystac.Item` instances
+            should be cloned before storing in the :class:`~ItemCollection`. Setting to
+            ``True`` ensures that changes made to :class:`~pystac.Item` instances in
+            the :class:`~ItemCollection` will not mutate the original ``Item``, but
+            will result in slower instantiation. Defaults to ``False``.
+
+    Examples:
+
+        Loop over all items in the :class`~ItemCollection`
+
+        >>> item_collection: ItemCollection = ...
+        >>> for item in item_collection:
+        ...     ...
+
+        Get the number of :class:`~pytac.Item` instances in the
+        :class:`~ItemCollection`
+
+        >>> length: int = len(item_collection)
+
+        Check if an :class:`~pystac.Item` is in the :class:`~ItemCollection`. Note
+        that the ``clone_items`` argument must be ``False`` for this to return
+        ``True``, since equality of PySTAC objects is currently evaluated using default
+        object equality (i.e. ``item_1 is item_2``).
+
+        >>> item: Item = ...
+        >>> item_collection = ItemCollection(items=[item])
+        >>> assert item in item_collection
+
+        Combine :class:`~ItemCollection` instances
+
+        >>> item_1: Item = ...
+        >>> item_2: Item = ...
+        >>> item_3: Item = ...
+        >>> item_collection_1 = ItemCollection(items=[item_1, item_2])
+        >>> item_collection_2 = ItemCollection(items=[item_2, item_3])
+        >>> combined = item_collection_1 + item_collection_2
+        >>> assert len(combined) == 3
+        # If an item is present in both ItemCollections it will only be added once
+    """
+
+    items: List[pystac.Item]
+    """List of :class:`pystac.Item` instances contained in this ``ItemCollection``."""
+
+    extra_fields: Dict[str, Any]
+    """Dictionary of additional top-level fields for the GeoJSON
+    FeatureCollection."""
+
+    def __init__(
+        self,
+        items: Iterable[ItemLike],
+        extra_fields: Optional[Dict[str, Any]] = None,
+        clone_items: bool = False,
+    ):
+        def map_item(item_or_dict: ItemLike) -> pystac.Item:
+            # Converts dicts to pystac.Items and clones if necessary
+            if isinstance(item_or_dict, pystac.Item):
+                return item_or_dict.clone() if clone_items else item_or_dict
+            else:
+                return pystac.Item.from_dict(item_or_dict)
+
+        self.items = list(map(map_item, items))
+        self.extra_fields = extra_fields or {}
+
+    def __getitem__(self, idx: int) -> pystac.Item:
+        return self.items[idx]
+
+    def __iter__(self) -> Iterator[pystac.Item]:
+        return iter(self.items)
+
+    def __len__(self) -> int:
+        return len(self.items)
+
+    def __contains__(self, __x: object) -> bool:
+        return __x in self.items
+
+    def __add__(self, other: object) -> "ItemCollection":
+        if not isinstance(other, ItemCollection):
+            return NotImplemented
+
+        combined = []
+        for item in self.items + other.items:
+            if item not in combined:
+                combined.append(item)
+
+        return ItemCollection(items=combined, clone_items=False)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Serializes an :class:`ItemCollection` instance to a JSON-like dictionary."""
+        return {
+            "type": "FeatureCollection",
+            "features": [item.to_dict() for item in self.items],
+            **self.extra_fields,
+        }
+
+    def clone(self) -> "ItemCollection":
+        """Creates a clone of this instance. This clone is a deep copy; all
+        :class:`~pystac.Item` instances are cloned and all additional top-level fields
+        are deep copied."""
+        return self.__class__(
+            items=[item.clone() for item in self.items],
+            extra_fields=deepcopy(self.extra_fields),
+        )
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "ItemCollection":
+        """Creates a :class:`ItemCollection` instance from a dictionary.
+
+        Arguments:
+            d : The dictionary from which the :class:`~ItemCollection` will be created
+        """
+        if not cls.is_item_collection(d):
+            raise STACTypeError("Dict is not a valid ItemCollection")
+
+        items = [pystac.Item.from_dict(item) for item in d.get("features", [])]
+        extra_fields = {k: v for k, v in d.items() if k not in ("features", "type")}
+
+        return cls(items=items, extra_fields=extra_fields)
+
+    @classmethod
+    def from_file(
+        cls, href: str, stac_io: Optional[pystac.StacIO] = None
+    ) -> "ItemCollection":
+        """Reads a :class:`ItemCollection` from a JSON file.
+
+        Arguments:
+            href : Path to the file.
+            stac_io : A :class:`~pystac.StacIO` instance to use for file I/O
+        """
+        if stac_io is None:
+            stac_io = pystac.StacIO.default()
+
+        if not is_absolute_href(href):
+            href = make_absolute_href(href)
+
+        d = stac_io.read_json(href)
+
+        return cls.from_dict(d)
+
+    def save_object(
+        self,
+        dest_href: str,
+        stac_io: Optional[pystac.StacIO] = None,
+    ) -> None:
+        """Saves this instance to the ``dest_href`` location.
+
+        Args:
+            dest_href : Location to which the file will be saved.
+            stac_io: Optional :class:`~pystac.StacIO` instance to use. If not provided,
+                will use the default instance.
+        """
+        if stac_io is None:
+            stac_io = pystac.StacIO.default()
+
+        stac_io.save_json(dest_href, self.to_dict())
+
+    @staticmethod
+    def is_item_collection(d: Dict[str, Any]) -> bool:
+        """Checks if the given dictionary represents a valid :class:`ItemCollection`.
+
+        Args:
+            d : Dictionary to check
+        """
+        typ = d.get("type")
+
+        # All ItemCollections are GeoJSON FeatureCollections
+        if typ != "FeatureCollection":
+            return False
+
+        # If it is a FeatureCollection and has a "stac_version" field, then it is an
+        #  ItemCollection. This will cover ItemCollections from STAC 0.9 to
+        #  <1.0.0-beta.1, when ItemCollections were removed from the core STAC Spec
+        if "stac_version" in d:
+            return True
+
+        # Prior to STAC 0.9 ItemCollections did not have a stac_version field and could
+        #  only be identified by the fact that all of their 'features' are STAC Items.
+        return all(
+            identify_stac_object_type(feature) == pystac.STACObjectType.ITEM
+            for feature in d.get("features", [])
+        )

pystac/serialization/__init__.py

@@ -51,4 +51,4 @@ def stac_object_from_dict(
     if info.object_type == pystac.STACObjectType.ITEM:
         return pystac.Item.from_dict(d, href=href, root=root, migrate=False)
 
-    raise ValueError(f"Unknown STAC object type {info.object_type}")
+    raise pystac.STACTypeError(f"Unknown STAC object type {info.object_type}")