Parse more datetime formats

stac_pydantic/api/search.py

@@ -11,11 +11,12 @@
     _GeometryBase,
 )
 from pydantic import BaseModel, Field, validator
+from pydantic.datetime_parse import parse_datetime
 
 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 DATETIME_RFC339, BBox
+from stac_pydantic.shared import BBox
 
 
 class Search(BaseModel):
@@ -42,16 +43,16 @@ def start_date(self) -> Optional[datetime]:
             return None
         if values[0] == "..":
             return None
-        return datetime.strptime(values[0], DATETIME_RFC339)
+        return parse_datetime(values[0])
 
     @property
     def end_date(self) -> Optional[datetime]:
         values = self.datetime.split("/")
         if len(values) == 1:
-            return datetime.strptime(values[0], DATETIME_RFC339)
+            return parse_datetime(values[0])
         if values[1] == "..":
             return None
-        return datetime.strptime(values[1], DATETIME_RFC339)
+        return parse_datetime(values[1])
 
     @validator("intersects")
     def validate_spatial(cls, v, values):
@@ -72,21 +73,16 @@ def validate_datetime(cls, v):
             if value == "..":
                 dates.append(value)
                 continue
-            try:
-                datetime.strptime(value, DATETIME_RFC339)
-                dates.append(value)
-            except:
-                raise ValueError(
-                    f"Invalid datetime, must match format ({DATETIME_RFC339})."
-                )
+
+            parse_datetime(value)
+            dates.append(value)
 
         if ".." not in dates:
-            if datetime.strptime(dates[0], DATETIME_RFC339) > datetime.strptime(
-                dates[1], DATETIME_RFC339
-            ):
+            if parse_datetime(dates[0]) > parse_datetime(dates[1]):
                 raise ValueError(
                     "Invalid datetime range, must match format (begin_date, end_date)"
                 )
+
         return v
 
     @property

stac_pydantic/item.py

@@ -4,6 +4,7 @@
 
 from geojson_pydantic.features import Feature, FeatureCollection
 from pydantic import BaseModel, Field, create_model, validator
+from pydantic.datetime_parse import parse_datetime
 from pydantic.fields import FieldInfo
 
 from stac_pydantic.api.extensions.context import ContextExtension
@@ -30,7 +31,7 @@ def validate_datetime(cls, v, values):
                 )
 
         if isinstance(v, str):
-            return cls._parse_rfc3339(v)
+            return parse_datetime(v)
 
         return v
 

stac_pydantic/shared.py

@@ -2,7 +2,7 @@
 from enum import Enum, auto
 from typing import List, Optional, Tuple, Union
 
-from pydantic import BaseModel, Extra, Field, validator
+from pydantic import BaseModel, Extra, Field
 
 from stac_pydantic.extensions.eo import BandObject
 from stac_pydantic.utils import AutoValueEnum
@@ -14,6 +14,7 @@
 ]
 
 # 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"
 
 
@@ -101,42 +102,17 @@ class StacCommonMetadata(BaseModel):
 
     title: Optional[str] = Field(None, alias="title")
     description: Optional[str] = Field(None, alias="description")
-    start_datetime: Optional[Union[datetime, str]] = Field(None, alias="start_datetime")
-    end_datetime: Optional[Union[datetime, str]] = Field(None, alias="end_datetime")
-    created: Optional[Union[datetime, str]] = Field(None, alias="created")
-    updated: Optional[Union[datetime, str]] = Field(None, alias="updated")
+    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")
     platform: Optional[str] = Field(None, alias="platform")
     instruments: Optional[List[str]] = Field(None, alias="instruments")
     constellation: Optional[str] = Field(None, alias="constellation")
     mission: Optional[str] = Field(None, alias="mission")
     providers: Optional[List[Provider]] = Field(None, alias="providers")
     gsd: Optional[NumType] = Field(None, alias="gsd")
 
-    @staticmethod
-    def _parse_rfc3339(dt: str):
-        try:
-            return datetime.strptime(dt, DATETIME_RFC339)
-        except Exception as e:
-            raise ValueError(
-                f"Invalid datetime, must match format ({DATETIME_RFC339})."
-            ) from e
-
-    @validator("start_datetime", allow_reuse=True)
-    def validate_start_datetime(cls, v):
-        return cls._parse_rfc3339(v)
-
-    @validator("end_datetime", allow_reuse=True)
-    def validate_start_datetime(cls, v):
-        return cls._parse_rfc3339(v)
-
-    @validator("created", allow_reuse=True)
-    def validate_start_datetime(cls, v):
-        return cls._parse_rfc3339(v)
-
-    @validator("updated", allow_reuse=True)
-    def validate_start_datetime(cls, v):
-        return cls._parse_rfc3339(v)
-
     class Config:
         json_encoders = {datetime: lambda v: v.strftime(DATETIME_RFC339)}
 

tests/conftest.py

@@ -7,8 +7,6 @@
 import requests
 from click.testing import CliRunner
 
-from stac_pydantic.shared import DATETIME_RFC339
-
 
 def request(url: str):
     r = requests.get(url)

tests/test_models.py

@@ -1,6 +1,6 @@
 import json
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytest
 from pydantic import BaseModel, Field, ValidationError
@@ -397,7 +397,7 @@ def test_invalid_spatial_search():
 
 def test_temporal_search_single_tailed():
     # Test single tailed
-    utcnow = datetime.utcnow().replace(microsecond=0)
+    utcnow = datetime.utcnow().replace(microsecond=0, tzinfo=timezone.utc)
     utcnow_str = utcnow.strftime(DATETIME_RFC339)
     search = Search(collections=["collection1"], datetime=utcnow_str)
     assert search.start_date == None
@@ -406,7 +406,7 @@ def test_temporal_search_single_tailed():
 
 def test_temporal_search_two_tailed():
     # Test two tailed
-    utcnow = datetime.utcnow().replace(microsecond=0)
+    utcnow = datetime.utcnow().replace(microsecond=0, tzinfo=timezone.utc)
     utcnow_str = utcnow.strftime(DATETIME_RFC339)
     search = Search(collections=["collection1"], datetime=f"{utcnow_str}/{utcnow_str}")
     assert search.start_date == search.end_date == utcnow
@@ -639,6 +639,12 @@ def test_validate_item_reraise_exception():
         validate_item(test_item, reraise_exception=True)
 
 
+def test_validate_item_rfc3339_with_partial_seconds():
+    test_item = request(EO_EXTENSION)
+    test_item["properties"]["updated"] = "2018-10-01T01:08:32.033Z"
+    assert validate_item(test_item)
+
+
 def test_multi_inheritance():
     test_item = request(EO_EXTENSION)