From 8a78f6759ccbc8d6a4ef64ea8698f599c9e4510e Mon Sep 17 00:00:00 2001 From: ksmuczynski Date: Mon, 1 Dec 2025 13:03:18 -0700 Subject: [PATCH 1/4] feat(schemas): add enum, depth and geometry validations and association create model - Add `schemas/validators.py` with `DepthIntervalMixin` for logical depth checks, `GeometryMixin` for WKT validation, and `validate_enum_input` for validation. - Add `CreateThingGeologicFormationAssociation` to `schemas_geologic_formation.py` to support creation of formation picks/stratigraphy. - Update `CreateGeologicFormation` to enforce uppercase formation codes and validate WKT boundaries. - Update `CreateAquiferSystem` to validate WKT boundaries. --- schemas/aquifer_system.py | 18 +++++++-- schemas/geologic_formation.py | 35 ++++++++++++++-- schemas/validators.py | 76 +++++++++++++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 6 deletions(-) create mode 100644 schemas/validators.py diff --git a/schemas/aquifer_system.py b/schemas/aquifer_system.py index 6c077aa90..3b0a58470 100644 --- a/schemas/aquifer_system.py +++ b/schemas/aquifer_system.py @@ -1,12 +1,14 @@ from typing import List -from pydantic import BaseModel +from pydantic import BaseModel, field_validator from schemas import BaseResponseModel +from schemas.validators import GeometryMixin, validate_enum_input +from core.enums import AquiferType, GeographicScale # Import specific Enums # ------ CREATE ---------- -class CreateAquiferSystem(BaseModel): +class CreateAquiferSystem(BaseModel, GeometryMixin): """ Schema for creating an aquifer system. Used during data transfer and API creation. @@ -16,7 +18,17 @@ class CreateAquiferSystem(BaseModel): description: str | None = None primary_aquifer_type: str geographic_scale: str - boundary: str | None = None + # boundary field inherited from GeometryMixin + + @field_validator("primary_aquifer_type", mode="before") + @classmethod + def check_aquifer_type(cls, v): + return validate_enum_input(v, AquiferType) + + @field_validator("geographic_scale", mode="before") + @classmethod + def check_geographic_scale(cls, v): + return validate_enum_input(v, GeographicScale) # ------ RESPONSE ---------- diff --git a/schemas/geologic_formation.py b/schemas/geologic_formation.py index 62e24ee8e..fbdaabb02 100644 --- a/schemas/geologic_formation.py +++ b/schemas/geologic_formation.py @@ -1,22 +1,51 @@ from typing import List -from pydantic import BaseModel +from pydantic import BaseModel, field_validator from schemas import BaseResponseModel +from schemas.validators import DepthIntervalMixin, GeometryMixin, validate_enum_input from core.enums import FormationCode, Lithology # ------ CREATE ---------- -class CreateGeologicFormation(BaseModel): +class CreateGeologicFormation(BaseModel, GeometryMixin): """ Schema for creating a geologic formation. Used during data transfer and API creation. """ + # formation_code has its own custom uppercase validator formation_code: FormationCode | None = None description: str | None = None lithology: Lithology | None = None - boundary: str | None = None + # boundary: inherited from GeometryMixin + + @field_validator("formation_code", mode="before") + @classmethod + def upper_case_code(cls, v: str | None) -> str | None: + """ + Automatically uppercase the formation code. + """ + if isinstance(v, str): + return v.upper() + return v + + @field_validator("lithology", mode="before") + @classmethod + def check_lithology(cls, v): + return validate_enum_input(v, Lithology) + + +class CreateThingGeologicFormationAssociation(BaseModel, DepthIntervalMixin): + """ + Schema for linking a Thing (Well) to a GeologicFormation. + Uses DepthIntervalMixin to enforce bottom_depth > top_depth. + """ + + thing_id: int + geologic_formation_id: int | None = None + top_depth: float + bottom_depth: float # ------ RESPONSE ---------- diff --git a/schemas/validators.py b/schemas/validators.py new file mode 100644 index 000000000..77e3f12dd --- /dev/null +++ b/schemas/validators.py @@ -0,0 +1,76 @@ +""" +schemas/validators.py +Reusable Pydantic validators and mixins for aquifer and geology related schemas. +May consider expansion for other domain models in the future. +""" + +from typing import Any, Type +from pydantic import model_validator, field_validator, BaseModel, ValueError + +from enum import Enum + + +def validate_enum_input(v: Any, enum_cls: Type[Enum]) -> Any: + """ + Validates that the input matches an enum value, either exactly or case-insensitively. + Returns the actual Enum member value. + """ + if v is None: + return None + + # 1. Check if it's already a valid enum member or value + try: + return enum_cls(v).value + except ValueError: + pass + + # 2. Case-insensitive fallback (for string inputs) + if isinstance(v, str): + v_lower = v.lower() + for member in enum_cls: + if str(member.value).lower() == v_lower: + return member.value + + # 3. Fail if no match found + valid_options = [str(e.value) for e in enum_cls] + raise ValueError(f"Invalid value '{v}'. Must be one of: {', '.join(valid_options)}") + + +class DepthIntervalMixin(BaseModel): + """ + Mixin to enforce that bottom_depth is greater than top_depth. + Assumes the model has 'top_depth' and 'bottom_depth' fields. + """ + + top_depth: float + bottom_depth: float + + @model_validator(mode="after") + def check_depth_logical_order(self) -> "DepthIntervalMixin": + if self.bottom_depth is not None and self.top_depth is not None: + if self.bottom_depth <= self.top_depth: + raise ValueError( + f"Bottom depth ({self.bottom_depth}) must be greater " + f"than top depth ({self.top_depth})" + ) + if self.top_depth < 0: + raise ValueError("Top depth cannot be negative.") + return self + + +class GeometryMixin(BaseModel): + """ + Mixin to validate WKT strings for boundary fields. + """ + + boundary: str | None = None + + @field_validator("boundary") + @classmethod + def validate_wkt(cls, v: str | None) -> str | None: + if v is None: + return v + + # Basic String Check + if not isinstance(v, str) or not v.strip(): + raise ValueError("Boundary must be a valid WKT string.") From a8c4afcaaea3c3aa1fa3ec35271ee142e10f8fb9 Mon Sep 17 00:00:00 2001 From: ksmuczynski Date: Tue, 2 Dec 2025 11:16:18 -0700 Subject: [PATCH 2/4] refactor(schemas): centralize validations and enforce strict enums - Delegate WKT boundary validation to `services/validation/geospatial.py` to enforce topological validity. - Update `CreateAquiferSystem` and `CreateGeologicFormation` to enforce strict Enum typing for controlled vocabularies, removing loose string coercion. --- schemas/aquifer_system.py | 23 ++++++--------------- schemas/geologic_formation.py | 9 ++------- schemas/validators.py | 38 +++-------------------------------- 3 files changed, 11 insertions(+), 59 deletions(-) diff --git a/schemas/aquifer_system.py b/schemas/aquifer_system.py index 3b0a58470..32ae680e6 100644 --- a/schemas/aquifer_system.py +++ b/schemas/aquifer_system.py @@ -1,9 +1,8 @@ from typing import List -from pydantic import BaseModel, field_validator - +from pydantic import BaseModel from schemas import BaseResponseModel -from schemas.validators import GeometryMixin, validate_enum_input +from schemas.validators import GeometryMixin from core.enums import AquiferType, GeographicScale # Import specific Enums @@ -16,20 +15,10 @@ class CreateAquiferSystem(BaseModel, GeometryMixin): name: str description: str | None = None - primary_aquifer_type: str - geographic_scale: str + primary_aquifer_type: AquiferType + geographic_scale: GeographicScale | None = None # boundary field inherited from GeometryMixin - @field_validator("primary_aquifer_type", mode="before") - @classmethod - def check_aquifer_type(cls, v): - return validate_enum_input(v, AquiferType) - - @field_validator("geographic_scale", mode="before") - @classmethod - def check_geographic_scale(cls, v): - return validate_enum_input(v, GeographicScale) - # ------ RESPONSE ---------- class GeoJSONGeometry(BaseModel): @@ -48,8 +37,8 @@ class GeoJSONProperties(BaseResponseModel): name: str description: str | None = None - primary_aquifer_type: str - geographic_scale: str + primary_aquifer_type: AquiferType + geographic_scale: GeographicScale | None = None class AquiferSystemGeoJSONResponse(BaseModel): diff --git a/schemas/geologic_formation.py b/schemas/geologic_formation.py index fbdaabb02..8ca452c4b 100644 --- a/schemas/geologic_formation.py +++ b/schemas/geologic_formation.py @@ -3,7 +3,7 @@ from pydantic import BaseModel, field_validator from schemas import BaseResponseModel -from schemas.validators import DepthIntervalMixin, GeometryMixin, validate_enum_input +from schemas.validators import DepthIntervalMixin, GeometryMixin from core.enums import FormationCode, Lithology @@ -30,11 +30,6 @@ def upper_case_code(cls, v: str | None) -> str | None: return v.upper() return v - @field_validator("lithology", mode="before") - @classmethod - def check_lithology(cls, v): - return validate_enum_input(v, Lithology) - class CreateThingGeologicFormationAssociation(BaseModel, DepthIntervalMixin): """ @@ -43,7 +38,7 @@ class CreateThingGeologicFormationAssociation(BaseModel, DepthIntervalMixin): """ thing_id: int - geologic_formation_id: int | None = None + geologic_formation_id: int top_depth: float bottom_depth: float diff --git a/schemas/validators.py b/schemas/validators.py index 77e3f12dd..61fdb47ae 100644 --- a/schemas/validators.py +++ b/schemas/validators.py @@ -4,36 +4,8 @@ May consider expansion for other domain models in the future. """ -from typing import Any, Type from pydantic import model_validator, field_validator, BaseModel, ValueError - -from enum import Enum - - -def validate_enum_input(v: Any, enum_cls: Type[Enum]) -> Any: - """ - Validates that the input matches an enum value, either exactly or case-insensitively. - Returns the actual Enum member value. - """ - if v is None: - return None - - # 1. Check if it's already a valid enum member or value - try: - return enum_cls(v).value - except ValueError: - pass - - # 2. Case-insensitive fallback (for string inputs) - if isinstance(v, str): - v_lower = v.lower() - for member in enum_cls: - if str(member.value).lower() == v_lower: - return member.value - - # 3. Fail if no match found - valid_options = [str(e.value) for e in enum_cls] - raise ValueError(f"Invalid value '{v}'. Must be one of: {', '.join(valid_options)}") +from services.validation.geospatial import validate_wkt_geometry class DepthIntervalMixin(BaseModel): @@ -61,6 +33,7 @@ def check_depth_logical_order(self) -> "DepthIntervalMixin": class GeometryMixin(BaseModel): """ Mixin to validate WKT strings for boundary fields. + Delegates logic to the validate_wkt_geometry service function. """ boundary: str | None = None @@ -68,9 +41,4 @@ class GeometryMixin(BaseModel): @field_validator("boundary") @classmethod def validate_wkt(cls, v: str | None) -> str | None: - if v is None: - return v - - # Basic String Check - if not isinstance(v, str) or not v.strip(): - raise ValueError("Boundary must be a valid WKT string.") + return validate_wkt_geometry(v) From 33ea2e2465fa9cdf4ab59b65e1dba5642ea090be Mon Sep 17 00:00:00 2001 From: ksmuczynski Date: Tue, 2 Dec 2025 11:19:35 -0700 Subject: [PATCH 3/4] refactor(schemas): remove None check on validators.DepthIntervalMixin `top_depth` and `bottom_depth` are required so the None check is redundant. Remove the None check. --- schemas/validators.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/schemas/validators.py b/schemas/validators.py index 61fdb47ae..bbbafee10 100644 --- a/schemas/validators.py +++ b/schemas/validators.py @@ -19,12 +19,11 @@ class DepthIntervalMixin(BaseModel): @model_validator(mode="after") def check_depth_logical_order(self) -> "DepthIntervalMixin": - if self.bottom_depth is not None and self.top_depth is not None: - if self.bottom_depth <= self.top_depth: - raise ValueError( - f"Bottom depth ({self.bottom_depth}) must be greater " - f"than top depth ({self.top_depth})" - ) + if self.bottom_depth <= self.top_depth: + raise ValueError( + f"Bottom depth ({self.bottom_depth}) must be greater " + f"than top depth ({self.top_depth})" + ) if self.top_depth < 0: raise ValueError("Top depth cannot be negative.") return self From 02107c3f9beee8baaba91550b623284a5a83fb29 Mon Sep 17 00:00:00 2001 From: ksmuczynski Date: Tue, 2 Dec 2025 11:37:27 -0700 Subject: [PATCH 4/4] refactor(schemas): enforce non-negative depths in DepthIntervalMixin - Remove manual `if` validation logic for non-negative depths in `DepthIntervalMixin`. - Implement `Field(ge=0)` on `top_depth` and `bottom_depth` to leverage Pydantic's native schema validation and cleaner OpenAPI generation. - Ensure `CreateThingGeologicFormationAssociation` inherits these constraints by explicitly redefining fields with `ge=0`. --- schemas/geologic_formation.py | 6 +++--- schemas/validators.py | 12 ++++++------ 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/schemas/geologic_formation.py b/schemas/geologic_formation.py index 8ca452c4b..de479bf29 100644 --- a/schemas/geologic_formation.py +++ b/schemas/geologic_formation.py @@ -1,6 +1,6 @@ from typing import List -from pydantic import BaseModel, field_validator +from pydantic import BaseModel, field_validator, Field from schemas import BaseResponseModel from schemas.validators import DepthIntervalMixin, GeometryMixin @@ -39,8 +39,8 @@ class CreateThingGeologicFormationAssociation(BaseModel, DepthIntervalMixin): thing_id: int geologic_formation_id: int - top_depth: float - bottom_depth: float + top_depth: float = Field(ge=0) + bottom_depth: float = Field(ge=0) # ------ RESPONSE ---------- diff --git a/schemas/validators.py b/schemas/validators.py index bbbafee10..fde828770 100644 --- a/schemas/validators.py +++ b/schemas/validators.py @@ -4,18 +4,20 @@ May consider expansion for other domain models in the future. """ -from pydantic import model_validator, field_validator, BaseModel, ValueError +from pydantic import model_validator, field_validator, BaseModel, ValueError, Field from services.validation.geospatial import validate_wkt_geometry class DepthIntervalMixin(BaseModel): """ - Mixin to enforce that bottom_depth is greater than top_depth. + Mixin to enforce: + 1. Depths are non-negative (via Field constraints). + 2. Bottom depth > top depth (via model_validator). Assumes the model has 'top_depth' and 'bottom_depth' fields. """ - top_depth: float - bottom_depth: float + top_depth: float = Field(ge=0) + bottom_depth: float = Field(ge=0) @model_validator(mode="after") def check_depth_logical_order(self) -> "DepthIntervalMixin": @@ -24,8 +26,6 @@ def check_depth_logical_order(self) -> "DepthIntervalMixin": f"Bottom depth ({self.bottom_depth}) must be greater " f"than top depth ({self.top_depth})" ) - if self.top_depth < 0: - raise ValueError("Top depth cannot be negative.") return self