Allow future extensions to be used (#1231)

* Allow future extensions to be used

* Pre-compile the regex

* Update changelog

---------

Co-authored-by: Pete Gadomski <pete.gadomski@gmail.com>

Author: jsignell

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Added
+
+- More permissive schema_uri matching to allow future versions of extension schemas ([#1231](https://github.com/stac-utils/pystac/pull/1231))
+
 ## [v1.8.4] - 2023-09-22
 
 ### Added

pystac/extensions/base.py

@@ -1,3 +1,5 @@
+import re
+import warnings
 from abc import ABC, abstractmethod
 from typing import (
     Any,
@@ -14,6 +16,8 @@
 
 import pystac
 
+VERSION_REGEX = re.compile("/v[0-9].[0-9].*/")
+
 
 class SummariesExtension:
     """Base class for extending the properties in :attr:`pystac.Collection.summaries`
@@ -115,6 +119,10 @@ def get_schema_uri(cls) -> str:
     @classmethod
     def get_schema_uris(cls) -> List[str]:
         """Gets a list of schema URIs associated with this extension."""
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return [cls.get_schema_uri()]
 
     @classmethod
@@ -140,8 +148,10 @@ def remove_from(cls, obj: S) -> None:
     def has_extension(cls, obj: S) -> bool:
         """Check if the given object implements this extension by checking
         :attr:`pystac.STACObject.stac_extensions` for this extension's schema URI."""
+        schema_startswith = re.split(VERSION_REGEX, cls.get_schema_uri())[0] + "/"
+
         return obj.stac_extensions is not None and any(
-            uri in obj.stac_extensions for uri in cls.get_schema_uris()
+            uri.startswith(schema_startswith) for uri in obj.stac_extensions
         )
 
     @classmethod

pystac/extensions/classification.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+import warnings
 from typing import (
     Any,
     Dict,
@@ -510,6 +511,10 @@ def get_schema_uri(cls) -> str:
 
     @classmethod
     def get_schema_uris(cls) -> List[str]:
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return [SCHEMA_URI_PATTERN.format(version=v) for v in SUPPORTED_VERSIONS]
 
     @classmethod
@@ -637,9 +642,11 @@ def bitfields(self, v: Optional[List[Bitfield]]) -> None:
 
 class ClassificationExtensionHooks(ExtensionHooks):
     schema_uri: str = SCHEMA_URI_PATTERN.format(version=DEFAULT_VERSION)
-    prev_extension_ids = set(ClassificationExtension.get_schema_uris()) - set(
-        [ClassificationExtension.get_schema_uri()]
-    )
+    prev_extension_ids = {
+        SCHEMA_URI_PATTERN.format(version=v)
+        for v in SUPPORTED_VERSIONS
+        if v != DEFAULT_VERSION
+    }
     stac_object_types = {pystac.STACObjectType.ITEM}
 
     def migrate(

pystac/extensions/eo.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import (
     Any,
     Dict,
@@ -385,6 +386,10 @@ def get_schema_uri(cls) -> str:
 
     @classmethod
     def get_schema_uris(cls) -> List[str]:
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return SCHEMA_URIS
 
     @classmethod

pystac/extensions/grid.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+import warnings
 from typing import Any, Dict, List, Optional, Pattern, Set, Union
 
 import pystac
@@ -86,6 +87,10 @@ def get_schema_uri(cls) -> str:
 
     @classmethod
     def get_schema_uris(cls) -> List[str]:
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return SCHEMA_URIS
 
     @classmethod

pystac/extensions/label.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import Any, Dict, Iterable, List, Optional, Sequence, Union, cast
 
 import pystac
@@ -696,6 +697,10 @@ def get_schema_uri(cls) -> str:
 
     @classmethod
     def get_schema_uris(cls) -> List[str]:
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return SCHEMA_URIS
 
     @classmethod

pystac/extensions/mgrs.py

@@ -8,6 +8,7 @@
 from pystac.extensions.hooks import ExtensionHooks
 
 SCHEMA_URI: str = "https://stac-extensions.github.io/mgrs/v1.0.0/schema.json"
+SCHEMA_STARTSWITH: str = "https://stac-extensions.github.io/mgrs/"
 PREFIX: str = "mgrs:"
 
 # Field names

pystac/extensions/projection.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import warnings
 from typing import Any, Dict, Generic, Iterable, List, Optional, TypeVar, Union, cast
 
 import pystac
@@ -265,6 +266,10 @@ def get_schema_uri(cls) -> str:
 
     @classmethod
     def get_schema_uris(cls) -> List[str]:
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return SCHEMA_URIS
 
     @classmethod

pystac/extensions/raster.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import (
     Any,
     Dict,
@@ -32,6 +33,7 @@
     "https://stac-extensions.github.io/raster/v1.0.0/schema.json",
     SCHEMA_URI,
 ]
+SCHEMA_STARTWITH = "https://stac-extensions.github.io/raster/"
 BANDS_PROP = "raster:bands"
 
 
@@ -715,6 +717,10 @@ def get_schema_uri(cls) -> str:
 
     @classmethod
     def get_schema_uris(cls) -> List[str]:
+        warnings.warn(
+            "get_schema_uris is deprecated and will be removed in v2",
+            DeprecationWarning,
+        )
         return SCHEMA_URIS
 
     @classmethod

tests/extensions/test_classification.py

@@ -94,12 +94,13 @@ def test_bitfield_object() -> None:
 
 
 def test_get_schema_uri(landsat_item: Item) -> None:
-    assert any(
-        [
-            uri in landsat_item.stac_extensions
-            for uri in ClassificationExtension.get_schema_uris()
-        ]
-    )
+    with pytest.raises(DeprecationWarning):
+        assert any(
+            [
+                uri in landsat_item.stac_extensions
+                for uri in ClassificationExtension.get_schema_uris()
+            ]
+        )
 
 
 def test_ext_raises_if_item_does_not_conform(plain_item: Item) -> None:

tests/extensions/test_projection.py

@@ -552,10 +552,10 @@ def test_summaries_adds_uri(self) -> None:
 
 def test_older_extension_version(projection_landsat8_item: Item) -> None:
     old = "https://stac-extensions.github.io/projection/v1.0.0/schema.json"
-    new = "https://stac-extensions.github.io/projection/v1.1.0/schema.json"
+    current = "https://stac-extensions.github.io/projection/v1.1.0/schema.json"
 
     stac_extensions = set(projection_landsat8_item.stac_extensions)
-    stac_extensions.remove(new)
+    stac_extensions.remove(current)
     stac_extensions.add(old)
     item_as_dict = projection_landsat8_item.to_dict(
         include_self_link=False, transform_hrefs=False
@@ -565,6 +565,26 @@ def test_older_extension_version(projection_landsat8_item: Item) -> None:
     assert ProjectionExtension.has_extension(item)
     assert old in item.stac_extensions
 
+    migrated_item = pystac.Item.from_dict(item_as_dict, migrate=True)
+    assert ProjectionExtension.has_extension(migrated_item)
+    assert current in migrated_item.stac_extensions
+
+
+def test_newer_extension_version(projection_landsat8_item: Item) -> None:
+    new = "https://stac-extensions.github.io/projection/v2.0.0/schema.json"
+    current = "https://stac-extensions.github.io/projection/v1.1.0/schema.json"
+
+    stac_extensions = set(projection_landsat8_item.stac_extensions)
+    stac_extensions.remove(current)
+    stac_extensions.add(new)
+    item_as_dict = projection_landsat8_item.to_dict(
+        include_self_link=False, transform_hrefs=False
+    )
+    item_as_dict["stac_extensions"] = list(stac_extensions)
+    item = Item.from_dict(item_as_dict)
+    assert ProjectionExtension.has_extension(item)
+    assert new in item.stac_extensions
+
     migrated_item = pystac.Item.from_dict(item_as_dict, migrate=True)
     assert ProjectionExtension.has_extension(migrated_item)
     assert new in migrated_item.stac_extensions