Migrate to Pydantic v2

CHANGELOG.txt

@@ -1,5 +1,6 @@
 Unreleased
 ------------------
+- Support pydantic>2.0 (@huard)
 
 2.0.3 (2022-5-3)
 ------------------
@@ -105,4 +106,4 @@ Unreleased
 - Allow extra asset-level fields (#1)
 - Fix population by field name model config, allowing model creation without extension namespaces (#2)
 - Add enum of commonly used asset media types (#3)
-- Move geojson models to `geojson-pydantic` library (#4)
+- Move geojson models to `geojson-pydantic` library (#4)

stac_pydantic/api/collections.py

@@ -15,7 +15,7 @@ class Collections(BaseModel):
     collections: List[Collection]
 
     def to_dict(self, **kwargs: Any) -> Dict[str, Any]:
-        return self.dict(by_alias=True, exclude_unset=True, **kwargs)
+        return self.model_dump(by_alias=True, exclude_unset=True, **kwargs)
 
     def to_json(self, **kwargs: Any) -> str:
-        return self.json(by_alias=True, exclude_unset=True, **kwargs)
+        return self.model_dump_json(by_alias=True, exclude_unset=True, **kwargs)

stac_pydantic/api/extensions/context.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict, Optional
 
-from pydantic import BaseModel, validator
+from pydantic import BaseModel, model_validator
 
 
 class ContextExtension(BaseModel):
@@ -9,12 +9,13 @@ class ContextExtension(BaseModel):
     """
 
     returned: int
-    limit: Optional[int]
-    matched: Optional[int]
+    limit: Optional[int] = None
+    matched: Optional[int] = None
 
-    @validator("limit")
-    def validate_limit(cls, v: Optional[int], values: Dict[str, Any]) -> None:
-        if values["returned"] > v:
+    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
+    @model_validator(mode="before")
+    def validate_limit(cls, values: Dict[str, Any]) -> None:
+        if values["limit"] and values["returned"] > values["limit"]:
             raise ValueError(
                 "Number of returned items must be less than or equal to the limit"
             )

stac_pydantic/api/extensions/fields.py

@@ -8,8 +8,8 @@ class FieldsExtension(BaseModel):
     https://github.com/radiantearth/stac-api-spec/tree/master/extensions/fields#fields-api-extension
     """
 
-    includes: Optional[Set[str]]
-    excludes: Optional[Set[str]]
+    includes: Optional[Set[str]] = None
+    excludes: Optional[Set[str]] = None
 
     def _get_field_dict(self, fields: Set[str]) -> Dict[str, Set[str]]:
         """Internal method to create a dictionary for advanced include or exclude of pydantic fields on model export

stac_pydantic/api/landing.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import Literal, List, Optional
 
 from pydantic import AnyUrl, BaseModel, Field
 
@@ -13,9 +13,9 @@ class LandingPage(BaseModel):
 
     id: str = Field(..., alias="id", min_length=1)
     description: str = Field(..., alias="description", min_length=1)
-    title: Optional[str]
-    stac_version: str = Field(STAC_VERSION, const=True)
-    stac_extensions: Optional[List[AnyUrl]]
+    title: Optional[str] = None
+    stac_version: Literal[STAC_VERSION] = STAC_VERSION
+    stac_extensions: Optional[List[AnyUrl]] = []
     conformsTo: List[AnyUrl]
     links: Links
-    type: str = Field("Catalog", const=True, min_length=1)
+    type: Literal["Catalog"] = "Catalog"

stac_pydantic/api/search.py

@@ -1,3 +1,4 @@
+import json
 from datetime import datetime as dt
 from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
@@ -11,8 +12,8 @@
     Polygon,
     _GeometryBase,
 )
-from pydantic import BaseModel, Field, validator
-from pydantic.datetime_parse import parse_datetime
+from pydantic import field_validator, BaseModel, Field, model_validator, TypeAdapter
+parse_datetime = lambda x: TypeAdapter(dt).validate_json(json.dumps(x))
 
 from stac_pydantic.api.extensions.fields import FieldsExtension
 from stac_pydantic.api.extensions.query import Operator
@@ -37,11 +38,11 @@ class Search(BaseModel):
     https://github.com/radiantearth/stac-api-spec/blob/master/api-spec.md#filter-parameters-and-fields
     """
 
-    collections: Optional[List[str]]
-    ids: Optional[List[str]]
-    bbox: Optional[BBox]
-    intersects: Optional[Intersection]
-    datetime: Optional[str]
+    collections: Optional[List[str]] = None
+    ids: Optional[List[str]] = None
+    bbox: Optional[BBox] = None
+    intersects: Optional[Intersection] = None
+    datetime: Optional[str] = None
     limit: int = 10
 
     @property
@@ -62,17 +63,15 @@ def end_date(self) -> Optional[dt]:
             return None
         return parse_datetime(values[1])
 
-    @validator("intersects")
-    def validate_spatial(
-        cls,
-        v: Intersection,
-        values: Dict[str, Any],
-    ) -> Intersection:
-        if v and values["bbox"] is not None:
+    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
+    @model_validator(mode="before")
+    def validate_spatial(cls, values: Dict[str, Any]) -> Intersection:
+        if values.get("intersects") and values.get("bbox") is not None:
             raise ValueError("intersects and bbox parameters are mutually exclusive")
-        return v
+        return values
 
-    @validator("bbox")
+    @field_validator("bbox")
+    @classmethod
     def validate_bbox(cls, v: BBox) -> BBox:
         if v:
             # Validate order
@@ -103,7 +102,8 @@ def validate_bbox(cls, v: BBox) -> BBox:
 
         return v
 
-    @validator("datetime")
+    @field_validator("datetime")
+    @classmethod
     def validate_datetime(cls, v: str) -> str:
         if "/" in v:
             values = v.split("/")
@@ -148,5 +148,5 @@ class ExtendedSearch(Search):
     """
 
     field: Optional[FieldsExtension] = Field(None, alias="fields")
-    query: Optional[Dict[str, Dict[Operator, Any]]]
-    sortby: Optional[List[SortExtension]]
+    query: Optional[Dict[str, Dict[Operator, Any]]] = None
+    sortby: Optional[List[SortExtension]] = None

stac_pydantic/catalog.py

@@ -1,6 +1,6 @@
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Literal
 
-from pydantic import AnyUrl, BaseModel, Field
+from pydantic import ConfigDict, AnyUrl, BaseModel, Field
 
 from stac_pydantic.links import Links
 from stac_pydantic.version import STAC_VERSION
@@ -11,20 +11,18 @@ class Catalog(BaseModel):
     https://github.com/radiantearth/stac-spec/blob/v1.0.0/catalog-spec/catalog-spec.md
     """
 
-    id: str = Field(..., alias="", min_length=1)
+    id: str = Field(..., alias="id", min_length=1)
     description: str = Field(..., alias="description", min_length=1)
-    stac_version: str = Field(STAC_VERSION, const=True, min_length=1)
+    stac_version: Literal[STAC_VERSION] = STAC_VERSION
     links: Links
-    stac_extensions: Optional[List[AnyUrl]]
-    title: Optional[str]
-    type: str = Field("Catalog", const=True, min_length=1)
+    stac_extensions: Optional[List[AnyUrl]] = []
+    title: Optional[str] = None
+    type: Literal["Catalog"] = "Catalog"
 
-    class Config:
-        use_enum_values = True
-        extra = "allow"
+    model_config = ConfigDict(use_enum_values=True, extra="allow")
 
     def to_dict(self: "Catalog", **kwargs: Any) -> Dict[str, Any]:
-        return self.dict(by_alias=True, exclude_unset=True, **kwargs)
+        return self.model_dump(by_alias=True, exclude_unset=True, **kwargs)
 
     def to_json(self: "Catalog", **kwargs: Any) -> str:
-        return self.json(by_alias=True, exclude_unset=True, **kwargs)
+        return self.model_dump_json(by_alias=True, exclude_unset=True, **kwargs)

stac_pydantic/collection.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union, Literal
 
 from pydantic import BaseModel, Field
 
@@ -45,11 +45,11 @@ class Collection(Catalog):
     https://github.com/radiantearth/stac-spec/blob/v1.0.0/collection-spec/collection-spec.md
     """
 
-    assets: Optional[Dict[str, Asset]]
+    assets: Optional[Dict[str, Asset]] = None
     license: str = Field(..., alias="license", min_length=1)
     extent: Extent
-    title: Optional[str]
-    keywords: Optional[List[str]]
-    providers: Optional[List[Provider]]
-    summaries: Optional[Dict[str, Union[Range, List[Any], Dict[str, Any]]]]
-    type: str = Field("Collection", const=True, min_length=1)
+    title: Optional[str] = None
+    keywords: Optional[List[str]] = None
+    providers: Optional[List[Provider]] = None
+    summaries: Optional[Dict[str, Union[Range, List[Any], Dict[str, Any]]]] = None
+    type: Literal["Collection"] = "Collection"

stac_pydantic/item.py

@@ -1,9 +1,10 @@
+import json
 from datetime import datetime as dt
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union, Literal
 
 from geojson_pydantic.features import Feature, FeatureCollection  # type: ignore
-from pydantic import AnyUrl, Field, root_validator, validator
-from pydantic.datetime_parse import parse_datetime
+from pydantic import model_validator, ConfigDict, AnyUrl, Field, field_validator, TypeAdapter, field_serializer
+parse_datetime = lambda x: TypeAdapter(dt).validate_json(json.dumps(x))
 
 from stac_pydantic.api.extensions.context import ContextExtension
 from stac_pydantic.links import Links
@@ -18,7 +19,8 @@ class ItemProperties(StacCommonMetadata):
 
     datetime: Union[dt, str] = Field(..., alias="datetime")
 
-    @validator("datetime")
+    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
+    @field_validator("datetime", mode="before")
     def validate_datetime(cls, v: Union[dt, str], values: Dict[str, Any]) -> dt:
         if v == "null":
             if not values["start_datetime"] and not values["end_datetime"]:
@@ -27,13 +29,16 @@ def validate_datetime(cls, v: Union[dt, str], values: Dict[str, Any]) -> dt:
                 )
 
         if isinstance(v, str):
-            return parse_datetime(v)
+            v = parse_datetime(v)
 
         return v
 
-    class Config:
-        extra = "allow"
-        json_encoders = {dt: lambda v: v.strftime(DATETIME_RFC339)}
+    @field_serializer("datetime")
+    def serialize_datetime(self, v: dt, _info):
+        return v.strftime(DATETIME_RFC339)
+
+    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-config for more information.
+    model_config = ConfigDict(extra="allow")
 
 
 class Item(Feature):  # type: ignore
@@ -42,23 +47,25 @@ class Item(Feature):  # type: ignore
     """
 
     id: str = Field(..., alias="id", min_length=1)
-    stac_version: str = Field(STAC_VERSION, const=True, min_length=1)
+    stac_version: Literal[STAC_VERSION] = STAC_VERSION
     properties: ItemProperties
     assets: Dict[str, Asset]
     links: Links
-    stac_extensions: Optional[List[AnyUrl]]
-    collection: Optional[str]
+    stac_extensions: Optional[List[AnyUrl]] = []
+    collection: Optional[str] = None
 
     def to_dict(self, **kwargs: Any) -> Dict[str, Any]:
-        return self.dict(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
+        return self.model_dump(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
 
     def to_json(self, **kwargs: Any) -> str:
-        return self.json(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
+        return self.model_dump_json(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
 
-    @root_validator(pre=True)
+    @model_validator(mode="before")
+    @classmethod
     def validate_bbox(cls, values: Dict[str, Any]) -> Dict[str, Any]:
-        if values.get("geometry") and values.get("bbox") is None:
-            raise ValueError("bbox is required if geometry is not null")
+        if isinstance(values, dict):
+            if values.get("geometry") and values.get("bbox") is None:
+                raise ValueError("bbox is required if geometry is not null")
         return values
 
 
@@ -67,14 +74,14 @@ class ItemCollection(FeatureCollection):  # type: ignore
     https://github.com/radiantearth/stac-spec/blob/v1.0.0/item-spec/itemcollection-spec.md
     """
 
-    stac_version: str = Field(STAC_VERSION, const=True, min_length=1)
+    stac_version: Literal[STAC_VERSION] = STAC_VERSION
     features: List[Item]
-    stac_extensions: Optional[List[AnyUrl]]
+    stac_extensions: Optional[List[AnyUrl]] = []
     links: Links
-    context: Optional[ContextExtension]
+    context: Optional[ContextExtension] = None
 
     def to_dict(self, **kwargs: Any) -> Dict[str, Any]:
-        return self.dict(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
+        return self.model_dump(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
 
     def to_json(self, **kwargs: Any) -> str:
-        return self.json(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore
+        return self.model_dump_json(by_alias=True, exclude_unset=True, **kwargs)  # type: ignore

stac_pydantic/links.py

@@ -2,7 +2,7 @@
 from typing import Any, Dict, Iterator, List, Optional, Union
 from urllib.parse import urljoin
 
-from pydantic import BaseModel, Field
+from pydantic import ConfigDict, BaseModel, Field, RootModel
 
 from stac_pydantic.utils import AutoValueEnum
 
@@ -32,13 +32,11 @@ class Link(BaseModel):
 
     href: str = Field(..., alias="href", min_length=1)
     rel: str = Field(..., alias="rel", min_length=1)
-    type: Optional[str]
-    title: Optional[str]
+    type: Optional[str] = None
+    title: Optional[str] = None
     # Label extension
     label: Optional[str] = Field(None, alias="label:assets")
-
-    class Config:
-        use_enum_values = True
+    model_config = ConfigDict(use_enum_values=True)
 
     def resolve(self, base_url: str) -> None:
         """resolve a link to the given base URL"""
@@ -52,30 +50,30 @@ class PaginationLink(Link):
 
     rel: PaginationRelations
     method: PaginationMethods
-    body: Optional[Dict[Any, Any]]
+    body: Optional[Dict[Any, Any]] = None
     merge: bool = False
 
 
-class Links(BaseModel):
-    __root__: List[Union[PaginationLink, Link]]
+class Links(RootModel):
+    root: List[Union[PaginationLink, Link]]
 
     def link_iterator(self) -> Iterator[Link]:
         """Produce iterator to iterate through links"""
-        return iter(self.__root__)
+        return iter(self.root)
 
     def resolve(self, base_url: str) -> None:
         """resolve all links to the given base URL"""
         for link in self.link_iterator():
             link.resolve(base_url)
 
     def append(self, link: Link) -> None:
-        self.__root__.append(link)
+        self.root.append(link)
 
     def __len__(self) -> int:
-        return len(self.__root__)
+        return len(self.root)
 
     def __getitem__(self, idx: int) -> Union[PaginationLink, Link]:
-        return self.__root__[idx]
+        return self.root[idx]
 
 
 class Relations(str, AutoValueEnum):