-
Couldn't load subscription status.
- Fork 30
move Search field validators outside model #174
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,15 +10,8 @@ | |
| Point, | ||
| Polygon, | ||
| ) | ||
| from pydantic import ( | ||
| BaseModel, | ||
| Field, | ||
| PrivateAttr, | ||
| TypeAdapter, | ||
| ValidationInfo, | ||
| field_validator, | ||
| model_validator, | ||
| ) | ||
| from pydantic import AfterValidator, BaseModel, Field, TypeAdapter, model_validator | ||
| from typing_extensions import Annotated | ||
|
|
||
| from stac_pydantic.api.extensions.fields import FieldsExtension | ||
| from stac_pydantic.api.extensions.query import Operator | ||
|
|
@@ -38,96 +31,65 @@ | |
| SearchDatetime = TypeAdapter(Optional[UtcDatetime]) | ||
|
|
||
|
|
||
| class Search(BaseModel): | ||
| """ | ||
| The base class for STAC API searches. | ||
| def validate_bbox(v: Optional[BBox]) -> Optional[BBox]: | ||
| """Validate BBOX value.""" | ||
| if v: | ||
| # Validate order | ||
| if len(v) == 4: | ||
| xmin, ymin, xmax, ymax = cast(Tuple[int, int, int, int], v) | ||
|
|
||
| https://github.com/radiantearth/stac-api-spec/blob/v1.0.0/item-search/README.md#query-parameter-table | ||
| """ | ||
| elif len(v) == 6: | ||
| xmin, ymin, min_elev, xmax, ymax, max_elev = cast( | ||
| Tuple[int, int, int, int, int, int], v | ||
| ) | ||
| if max_elev < min_elev: | ||
| raise ValueError( | ||
| "Maximum elevation must greater than minimum elevation" | ||
| ) | ||
| else: | ||
| raise ValueError("Bounding box must have 4 or 6 coordinates") | ||
|
|
||
| collections: Optional[List[str]] = None | ||
| ids: Optional[List[str]] = None | ||
| bbox: Optional[BBox] = None | ||
| intersects: Optional[Intersection] = None | ||
| datetime: Optional[str] = None | ||
| limit: Optional[int] = 10 | ||
| # Validate against WGS84 | ||
| if xmin < -180 or ymin < -90 or xmax > 180 or ymax > 90: | ||
| raise ValueError("Bounding box must be within (-180, -90, 180, 90)") | ||
|
|
||
| # Private properties to store the parsed datetime values. Not part of the model schema. | ||
| _start_date: Optional[dt] = PrivateAttr(default=None) | ||
| _end_date: Optional[dt] = PrivateAttr(default=None) | ||
| if ymax < ymin: | ||
| raise ValueError("Maximum longitude must be greater than minimum longitude") | ||
vincentsarago marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
|
|
||
| # Properties to return the private values | ||
| @property | ||
| def start_date(self) -> Optional[dt]: | ||
| return self._start_date | ||
| return v | ||
|
|
||
| @property | ||
| def end_date(self) -> Optional[dt]: | ||
| return self._end_date | ||
|
|
||
| # 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]) -> Dict[str, Any]: | ||
| if values.get("intersects") and values.get("bbox") is not None: | ||
| raise ValueError("intersects and bbox parameters are mutually exclusive") | ||
| return values | ||
| def str_to_datetimes(value: str) -> List[Optional[dt]]: | ||
gadomski marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| # Split on "/" and replace no value or ".." with None | ||
| values = [v if v and v != ".." else None for v in value.split("/")] | ||
|
|
||
| @field_validator("bbox") | ||
| @classmethod | ||
| def validate_bbox(cls, v: BBox) -> BBox: | ||
| if v: | ||
| # Validate order | ||
| if len(v) == 4: | ||
| xmin, ymin, xmax, ymax = cast(Tuple[int, int, int, int], v) | ||
| else: | ||
| xmin, ymin, min_elev, xmax, ymax, max_elev = cast( | ||
| Tuple[int, int, int, int, int, int], v | ||
| ) | ||
| if max_elev < min_elev: | ||
| raise ValueError( | ||
| "Maximum elevation must greater than minimum elevation" | ||
| ) | ||
| # Validate against WGS84 | ||
| if xmin < -180 or ymin < -90 or xmax > 180 or ymax > 90: | ||
| raise ValueError("Bounding box must be within (-180, -90, 180, 90)") | ||
|
|
||
| if ymax < ymin: | ||
| raise ValueError( | ||
| "Maximum longitude must be greater than minimum longitude" | ||
| ) | ||
| # 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 | ||
| ], | ||
| ) | ||
| return dates | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. extract the |
||
|
|
||
| return v | ||
|
|
||
| @field_validator("datetime", mode="after") | ||
| @classmethod | ||
| def validate_datetime( | ||
| cls, value: Optional[str], info: ValidationInfo | ||
| ) -> Optional[str]: | ||
| # Split on "/" and replace no value or ".." with None | ||
| if value is None: | ||
| return value | ||
| values = [v if v and v != ".." else None for v in value.split("/")] | ||
| def validate_datetime(v: Optional[str]) -> Optional[str]: | ||
| """Validate Datetime value.""" | ||
| if v is not None: | ||
| dates = str_to_datetimes(v) | ||
|
|
||
| # If there are more than 2 dates, it's invalid | ||
| if len(values) > 2: | ||
| if len(dates) > 2: | ||
| raise ValueError( | ||
| "Invalid datetime range. Too many values. Must match format: {begin_date}/{end_date}" | ||
| ) | ||
|
|
||
| # If there is only one date, duplicate to use for both start and end dates | ||
| if len(values) == 1: | ||
| values = [values[0], values[0]] | ||
|
|
||
| # 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 len(dates) == 1: | ||
| dates = [dates[0], dates[0]] | ||
|
|
||
| # 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]: | ||
|
|
@@ -136,12 +98,44 @@ def validate_datetime( | |
| "Must match format: {begin_date}/{end_date}" | ||
| ) | ||
|
|
||
| # Store the parsed dates | ||
| info.data["_start_date"] = dates[0] | ||
| info.data["_end_date"] = dates[1] | ||
| return v | ||
|
|
||
|
|
||
| class Search(BaseModel): | ||
| """ | ||
| The base class for STAC API searches. | ||
|
|
||
| https://github.com/radiantearth/stac-api-spec/blob/v1.0.0/item-search/README.md#query-parameter-table | ||
| """ | ||
|
|
||
| collections: Optional[List[str]] = None | ||
| ids: Optional[List[str]] = None | ||
| bbox: Annotated[Optional[BBox], AfterValidator(validate_bbox)] = None | ||
| intersects: Optional[Intersection] = None | ||
| datetime: Annotated[Optional[str], AfterValidator(validate_datetime)] = None | ||
| limit: Optional[int] = 10 | ||
|
|
||
| @property | ||
| def start_date(self) -> Optional[dt]: | ||
| start_date: Optional[dt] = None | ||
| if self.datetime: | ||
| start_date = str_to_datetimes(self.datetime)[0] | ||
| return start_date | ||
|
|
||
| @property | ||
| def end_date(self) -> Optional[dt]: | ||
| end_date: Optional[dt] = None | ||
| if self.datetime: | ||
| dates = str_to_datetimes(self.datetime) | ||
| end_date = dates[0] if len(dates) == 1 else dates[1] | ||
| return end_date | ||
|
|
||
| # Return the original string value | ||
| return value | ||
| # 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]) -> Dict[str, Any]: | ||
| if values.get("intersects") and values.get("bbox") is not None: | ||
| raise ValueError("intersects and bbox parameters are mutually exclusive") | ||
| return values | ||
|
|
||
| @property | ||
| def spatial_filter(self) -> Optional[Intersection]: | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This will not be called normally because pydantic will already check if the inputs are tuple of 4 or 6 values. But I think if we want to re-use the validator outside pydantic model (e.g in stac-fastapi) it will be useful