Update datetime strategy.

Author: eseglem

pyproject.toml

@@ -19,7 +19,7 @@ keywords=["stac", "pydantic", "validation"]
 authors=[{ name = "Arturo Engineering", email = "[email protected]"}]
 license= { text = "MIT" }
 requires-python=">=3.8"
-dependencies = ["click>=8.1.7", "pydantic>=2.4.1", "geojson-pydantic>=1.0.0", "ciso8601~=2.3"]
+dependencies = ["click>=8.1.7", "pydantic>=2.4.1", "geojson-pydantic>=1.0.0"]
 dynamic = ["version", "readme"]
 
 [project.scripts]

stac_pydantic/api/search.py

@@ -1,8 +1,7 @@
 from datetime import datetime as dt
 from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
-from ciso8601 import parse_rfc3339
-from geojson_pydantic.geometries import (  # type: ignore
+from geojson_pydantic.geometries import (
     GeometryCollection,
     LineString,
     MultiLineString,
@@ -11,12 +10,12 @@
     Point,
     Polygon,
 )
-from pydantic import BaseModel, Field, field_validator, model_validator
+from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_validator
 
 from stac_pydantic.api.extensions.fields import FieldsExtension
 from stac_pydantic.api.extensions.query import Operator
 from stac_pydantic.api.extensions.sort import SortExtension
-from stac_pydantic.shared import BBox
+from stac_pydantic.shared import BBox, UtcDatetime
 
 Intersection = Union[
     Point,
@@ -28,6 +27,8 @@
     GeometryCollection,
 ]
 
+SearchDatetime = TypeAdapter(Optional[UtcDatetime])
+
 
 class Search(BaseModel):
     """
@@ -43,23 +44,18 @@ class Search(BaseModel):
     datetime: Optional[str] = None
     limit: int = 10
 
+    # Private properties to store the parsed datetime values. Not part of the model schema.
+    _start_date: Optional[dt] = None
+    _end_date: Optional[dt] = None
+
+    # Properties to return the private values
     @property
     def start_date(self) -> Optional[dt]:
-        values = (self.datetime or "").split("/")
-        if len(values) == 1:
-            return None
-        if values[0] == ".." or values[0] == "":
-            return None
-        return parse_rfc3339(values[0])
+        return self._start_date
 
     @property
     def end_date(self) -> Optional[dt]:
-        values = (self.datetime or "").split("/")
-        if len(values) == 1:
-            return parse_rfc3339(values[0])
-        if values[1] == ".." or values[1] == "":
-            return None
-        return parse_rfc3339(values[1])
+        return self._end_date
 
     # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
     @model_validator(mode="before")
@@ -102,30 +98,38 @@ def validate_bbox(cls, v: BBox) -> BBox:
 
     @field_validator("datetime")
     @classmethod
-    def validate_datetime(cls, v: str) -> str:
-        if "/" in v:
-            values = v.split("/")
-        else:
-            # Single date is interpreted as end date
-            values = ["..", v]
-
-        dates: List[dt] = []
-        for value in values:
-            if value == ".." or value == "":
-                continue
-
-            dates.append(parse_rfc3339(value))
-
+    def validate_datetime(cls, value: str) -> str:
+        # Split on "/" and replace no value or ".." with None
+        values = [v if v and v != ".." else None for v in value.split("/")]
+        # If there are more than 2 dates, it's invalid
         if len(values) > 2:
-            raise ValueError("Invalid datetime range, must match format (begin_date, end_date)")
-
-        if not {"..", ""}.intersection(set(values)):
-            if dates[0] > dates[1]:
-                raise ValueError(
-                    "Invalid datetime range, must match format (begin_date, end_date)"
-                )
-
-        return v
+            raise ValueError(
+                "Invalid datetime range. Too many values. Must match format: {begin_date}/{end_date}"
+            )
+        # If there is only one date, insert a None for the start date
+        if len(values) == 1:
+            values.insert(0, None)
+        # Cast because pylance gets confused by the type adapter and annotated type
+        dates = cast(
+            List[Optional[dt]],
+            [
+                # Use the type adapter to validate the datetime strings, strict is necessary
+                # due to pydantic issues #8736 and #8762
+                SearchDatetime.validate_strings(v, strict=True) if v else None
+                for v in values
+            ],
+        )
+        # If there is a start and end date, check that the start date is before the end date
+        if dates[0] and dates[1] and dates[0] > dates[1]:
+            raise ValueError(
+                "Invalid datetime range. Begin date after end date. "
+                "Must match format: {begin_date}/{end_date}"
+            )
+        # Store the parsed dates
+        cls._start_date = dates[0]
+        cls._end_date = dates[1]
+        # Return the original string value
+        return value
 
     @property
     def spatial_filter(self) -> Optional[Intersection]:

stac_pydantic/item.py

@@ -1,4 +1,3 @@
-from datetime import datetime as dt
 from typing import Any, Dict, List, Optional
 
 from geojson_pydantic import Feature
@@ -9,13 +8,15 @@
     model_serializer,
     model_validator,
 )
+from typing_extensions import Self
 
 from stac_pydantic.links import Links
 from stac_pydantic.shared import (
     SEMVER_REGEX,
     Asset,
     StacBaseModel,
     StacCommonMetadata,
+    UtcDatetime,
 )
 from stac_pydantic.version import STAC_VERSION
 
@@ -25,13 +26,13 @@ class ItemProperties(StacCommonMetadata):
     https://github.com/radiantearth/stac-spec/blob/v1.0.0/item-spec/item-spec.md#properties-object
     """
 
-    datetime: Optional[dt] = Field(..., alias="datetime")
+    datetime: Optional[UtcDatetime]
 
     # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-config for more information.
     model_config = ConfigDict(extra="allow")
 
     @model_validator(mode="after")
-    def validate_datetime(self) -> "ItemProperties":
+    def validate_datetime(self) -> Self:
         if not self.datetime and (not self.start_datetime or not self.end_datetime):
             raise ValueError(
                 "start_datetime and end_datetime must be specified when datetime is null"

stac_pydantic/shared.py

@@ -1,9 +1,16 @@
-from datetime import datetime
+from datetime import timezone
 from enum import Enum, auto
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Annotated, Any, Dict, List, Optional, Tuple, Union
 from warnings import warn
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import (
+    AfterValidator,
+    AwareDatetime,
+    BaseModel,
+    ConfigDict,
+    Field,
+    PlainSerializer,
+)
 
 from stac_pydantic.utils import AutoValueEnum
 
@@ -15,9 +22,16 @@
 
 SEMVER_REGEX = r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$"
 
-# https://tools.ietf.org/html/rfc3339#section-5.6
-# Unused, but leaving it here since it's used by dependencies
-DATETIME_RFC339 = "%Y-%m-%dT%H:%M:%SZ"
+# Allows for some additional flexibility in the input datetime format. As long as
+# the input value has timezone information, it will be converted to UTC timezone.
+UtcDatetime = Annotated[
+    # Input value must be in a format which has timezone information
+    AwareDatetime,
+    # Convert the input value to UTC timezone
+    AfterValidator(lambda d: d.astimezone(timezone.utc)),
+    # Use `isoformat` to serialize the value in an RFC3339 compatible format
+    PlainSerializer(lambda d: d.isoformat()),
+]
 
 
 class MimeTypes(str, Enum):
@@ -118,10 +132,10 @@ class StacCommonMetadata(StacBaseModel):
 
     title: Optional[str] = Field(None, alias="title")
     description: Optional[str] = Field(None, alias="description")
-    start_datetime: Optional[datetime] = Field(None, alias="start_datetime")
-    end_datetime: Optional[datetime] = Field(None, alias="end_datetime")
-    created: Optional[datetime] = Field(None, alias="created")
-    updated: Optional[datetime] = Field(None, alias="updated")
+    start_datetime: Optional[UtcDatetime] = Field(None, alias="start_datetime")
+    end_datetime: Optional[UtcDatetime] = Field(None, alias="end_datetime")
+    created: Optional[UtcDatetime] = Field(None, alias="created")
+    updated: Optional[UtcDatetime] = Field(None, alias="updated")
     platform: Optional[str] = Field(None, alias="platform")
     instruments: Optional[List[str]] = Field(None, alias="instruments")
     constellation: Optional[str] = Field(None, alias="constellation")

tests/api/extensions/test_fields.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 
 from shapely.geometry import Polygon
 
@@ -15,7 +15,7 @@ def test_fields_filter_item():
     item = Item(
         id="test-fields-filter",
         geometry=Polygon.from_bounds(0, 0, 0, 0),
-        properties={"datetime": datetime.utcnow(), "foo": "foo", "bar": "bar"},
+        properties={"datetime": datetime.now(timezone.utc), "foo": "foo", "bar": "bar"},
         assets={},
         links=[
             {"href": "http://link", "rel": "self"},

tests/api/test_search.py

@@ -1,12 +1,11 @@
 import time
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timedelta, timezone
 
 import pytest
 from pydantic import ValidationError
 from shapely.geometry import Polygon, shape
 
 from stac_pydantic.api.search import Search
-from stac_pydantic.shared import DATETIME_RFC339
 
 
 def test_search():
@@ -57,17 +56,17 @@ def test_invalid_spatial_search():
 
 def test_temporal_search_single_tailed():
     # Test single tailed
-    utcnow = datetime.utcnow().replace(microsecond=0, tzinfo=timezone.utc)
-    utcnow_str = utcnow.strftime(DATETIME_RFC339)
+    utcnow = datetime.now(timezone.utc).replace(microsecond=0, tzinfo=timezone.utc)
+    utcnow_str = utcnow.isoformat()
     search = Search(collections=["collection1"], datetime=utcnow_str)
     assert search.start_date is None
     assert search.end_date == utcnow
 
 
 def test_temporal_search_two_tailed():
     # Test two tailed
-    utcnow = datetime.utcnow().replace(microsecond=0, tzinfo=timezone.utc)
-    utcnow_str = utcnow.strftime(DATETIME_RFC339)
+    utcnow = datetime.now(timezone.utc).replace(microsecond=0, tzinfo=timezone.utc)
+    utcnow_str = utcnow.isoformat()
     search = Search(collections=["collection1"], datetime=f"{utcnow_str}/{utcnow_str}")
     assert search.start_date == search.end_date == utcnow
 
@@ -87,26 +86,34 @@ def test_temporal_search_open():
     assert search.end_date is None
 
 
-def test_invalid_temporal_search():
-    # Not RFC339
-    utcnow = datetime.utcnow().strftime("%Y-%m-%d")
+def test_invalid_temporal_search_date():
+    # Just a date, no time
+    utcnow = datetime.now(timezone.utc).strftime("%Y-%m-%d")
     with pytest.raises(ValidationError):
         Search(collections=["collection1"], datetime=utcnow)
 
-    t1 = datetime.utcnow()
+
+def test_invalid_temporal_search_too_many():
+    # Too many dates
+    t1 = datetime.now(timezone.utc)
     t2 = t1 + timedelta(seconds=100)
     t3 = t2 + timedelta(seconds=100)
     with pytest.raises(ValidationError):
-        Search(collections=["collection1"], datetime=f"{t1.strftime(DATETIME_RFC339)}/{t2.strftime(DATETIME_RFC339)}/{t3.strftime(DATETIME_RFC339)}",)
+        Search(
+            collections=["collection1"],
+            datetime=f"{t1.isoformat()}/{t2.isoformat()}/{t3.isoformat()}",
+        )
+
 
+def test_invalid_temporal_search_date_wrong_order():
     # End date is before start date
-    start = datetime.utcnow()
+    start = datetime.now(timezone.utc)
     time.sleep(2)
-    end = datetime.utcnow()
+    end = datetime.now(timezone.utc)
     with pytest.raises(ValidationError):
         Search(
             collections=["collection1"],
-            datetime=f"{end.strftime(DATETIME_RFC339)}/{start.strftime(DATETIME_RFC339)}",
+            datetime=f"{end.isoformat()}/{start.isoformat()}",
         )