fix: tighten pydantic validation contracts

Author: ruicore

src/excelalchemy/adapters/pydantic.py

@@ -132,6 +132,65 @@ def _normalize_validation_message(
     return NormalizedValidationMessage(normalized)
 
 
+def _normalize_pydantic_error(
+    error: Mapping[str, object],
+    field_def: FieldMetaInfo | None = None,
+    *,
+    excel_codec: type[ExcelFieldCodec] | None = None,
+) -> NormalizedValidationMessage:
+    normalized_message = _normalize_constraint_error(error, field_def, excel_codec=excel_codec)
+    if normalized_message is not None:
+        return normalized_message
+    return _normalize_validation_message(error.get('msg', ''), field_def, excel_codec=excel_codec)
+
+
+def _normalize_constraint_error(
+    error: Mapping[str, object],
+    field_def: FieldMetaInfo | None,
+    *,
+    excel_codec: type[ExcelFieldCodec] | None = None,
+) -> NormalizedValidationMessage | None:
+    if field_def is None:
+        return None
+
+    error_type = error.get('type')
+    ctx = error.get('ctx')
+    if not isinstance(ctx, Mapping):
+        return None
+
+    ctx = cast(Mapping[object, object], ctx)
+    field_type = ctx.get('field_type')
+    constraints = field_def.constraints
+    if error_type == 'too_short' and field_type == 'List':
+        min_items = _int_from_context(ctx, 'min_length') or constraints.min_items
+        if min_items is not None:
+            return NormalizedValidationMessage.from_key(MessageKey.MIN_ITEMS_REQUIRED, min_items=min_items)
+
+    if error_type == 'too_long' and field_type == 'List':
+        max_items = _int_from_context(ctx, 'max_length') or constraints.max_items
+        if max_items is not None:
+            return NormalizedValidationMessage.from_key(MessageKey.MAX_ITEMS_ALLOWED, max_items=max_items)
+
+    if error_type == 'string_too_short':
+        min_length = _int_from_context(ctx, 'min_length') or constraints.min_length
+        if min_length is not None:
+            return NormalizedValidationMessage.from_key(MessageKey.MIN_LENGTH_CHARACTERS, min_length=min_length)
+
+    if error_type == 'string_too_long':
+        max_length = _int_from_context(ctx, 'max_length') or constraints.max_length
+        if max_length is not None:
+            return NormalizedValidationMessage.from_key(MessageKey.MAX_LENGTH_CHARACTERS, max_length=max_length)
+
+    return _normalize_constraint_message(str(error.get('msg', '')), field_def, excel_codec=excel_codec)
+
+
+def _int_from_context(ctx: Mapping[object, object], key: str) -> int | None:
+    value = ctx.get(key)
+    if isinstance(value, int):
+        return value
+    return None
+
+
 def _normalize_constraint_message(
     message: str,
     field_def: FieldMetaInfo | None,
@@ -293,10 +352,50 @@ def fields(self) -> Iterable[PydanticFieldAdapter]:
     def field(self, name: str) -> PydanticFieldAdapter:
         return PydanticFieldAdapter(name=name, raw_field=self.model.model_fields[name])
 
+    def field_for_validation_location(self, location: str) -> PydanticFieldAdapter | None:
+        field_name = self._field_name_for_validation_location(location)
+        if field_name is None:
+            return None
+        return self.field(field_name)
+
+    def _field_name_for_validation_location(self, location: str) -> str | None:
+        if location in self.model.model_fields:
+            return location
+
+        for name, field_info in self.model.model_fields.items():
+            if location in _validation_locations_for_field(field_info):
+                return name
+
+        return None
+
     def field_names(self) -> list[str]:
         return list(self.model.model_fields.keys())
 
 
+def _validation_locations_for_field(field_info: FieldInfo) -> set[str]:
+    locations: set[str] = set()
+    if isinstance(field_info.alias, str):
+        locations.add(field_info.alias)
+    _collect_validation_alias_locations(field_info.validation_alias, locations)
+    return locations
+
+
+def _collect_validation_alias_locations(alias: object, locations: set[str]) -> None:
+    if isinstance(alias, str):
+        locations.add(alias)
+        return
+
+    choices = getattr(alias, 'choices', None)
+    if isinstance(choices, (list, tuple)):
+        for choice in cast(Iterable[object], choices):
+            _collect_validation_alias_locations(choice, locations)
+        return
+
+    path = getattr(alias, 'path', None)
+    if isinstance(path, (list, tuple)) and path and isinstance(path[0], str):
+        locations.add(path[0])
+
+
 def extract_pydantic_model(
     model: type[BaseModel] | None,
 ) -> list[FieldMetaInfo]:
@@ -383,7 +482,7 @@ def _model_validate[ModelT: BaseModel](
     failed_fields: set[str],
 ) -> ModelT | list[ExcelCellError | ExcelRowError]:
     try:
-        return model.model_validate(data)
+        return model.model_validate(data, by_alias=False, by_name=True)
     except ValidationError as exc:
         return _map_validation_error(exc, model_adapter, failed_fields)
 
@@ -397,21 +496,26 @@ def _map_validation_error(
     for error in exc.errors():
         loc = error.get('loc', ())
         if not loc:
-            normalized = _normalize_validation_message(str(error['msg']))
+            normalized = _normalize_pydantic_error(error)
             mapped.append(_build_row_error(normalized))
             continue
 
         field_name = loc[0]
         if not isinstance(field_name, str):
-            normalized = _normalize_validation_message(str(error['msg']))
+            normalized = _normalize_pydantic_error(error)
             mapped.append(_build_row_error(normalized))
             continue
         if field_name in failed_fields:
             continue
 
-        field_adapter = model_adapter.field(field_name)
-        normalized = _normalize_validation_message(
-            str(error['msg']),
+        field_adapter = model_adapter.field_for_validation_location(field_name)
+        if field_adapter is None:
+            normalized = _normalize_pydantic_error(error)
+            mapped.append(_build_row_error(normalized))
+            continue
+
+        normalized = _normalize_pydantic_error(
+            error,
             field_adapter.declared_metadata,
             excel_codec=field_adapter.excel_codec,
         )

src/excelalchemy/results/import_result.py

@@ -22,29 +22,21 @@ class ValidateResult(StrEnum):
 class ImportResult(BaseModel):
     """Structured result returned from an import run."""
 
-    model_config = ConfigDict(extra='allow')
+    # fmt: off
+    model_config = ConfigDict(extra='forbid')
 
     result: ValidateResult = Field(description='Overall import result.')
 
     is_required_missing: bool = Field(default=False, description='Whether required headers are missing.')
-    missing_required: list[Label] = Field(
-        default_factory=empty_labels, description='Required headers missing from the workbook.'
-    )
-    missing_primary: list[Label] = Field(
-        default_factory=empty_labels, description='Primary-key headers missing from the workbook.'
-    )
-    unrecognized: list[Label] = Field(
-        default_factory=empty_labels, description='Headers present in the workbook but unknown to the schema.'
-    )
-    duplicated: list[Label] = Field(
-        default_factory=empty_labels, description='Headers that appear more than once in the workbook.'
-    )
-
-    url: str | None = Field(
-        default=None, description='Download URL for the import result workbook when one is produced.'
-    )
+    missing_required: list[Label] = Field(default_factory=empty_labels, description='Required headers missing from the workbook.')
+    missing_primary: list[Label] = Field(default_factory=empty_labels, description='Primary-key headers missing from the workbook.')
+    unrecognized: list[Label] = Field(default_factory=empty_labels, description='Headers present in the workbook but unknown to the schema.')
+    duplicated: list[Label] = Field(default_factory=empty_labels, description='Headers that appear more than once in the workbook.')
+
+    url: str | None = Field(default=None, description='Download URL for the import result workbook when one is produced.')
     success_count: int = Field(default=0, description='Number of rows imported successfully.')
     fail_count: int = Field(default=0, description='Number of rows that failed to import.')
+    # fmt: on
 
     @property
     def is_success(self) -> bool:

src/excelalchemy/results/preflight.py

@@ -14,11 +14,13 @@ def empty_labels() -> list[Label]:
 class ValidateHeaderResult(BaseModel):
     """Header validation result."""
 
+    # fmt: off
     missing_required: list[Label] = Field(description='Required headers missing from the workbook.')
     missing_primary: list[Label] = Field(description='Primary-key headers missing from the workbook.')
     unrecognized: list[Label] = Field(description='Headers present in the workbook but unknown to the schema.')
     duplicated: list[Label] = Field(description='Headers that appear more than once in the workbook.')
     is_valid: bool = Field(default=True, description='Whether header validation succeeded.')
+    # fmt: on
 
     @property
     def is_required_missing(self) -> bool:
@@ -38,32 +40,20 @@ class ImportPreflightStatus(StrEnum):
 class ImportPreflightResult(BaseModel):
     """Structured result returned from lightweight import preflight."""
 
+    # fmt: off
     status: ImportPreflightStatus = Field(description='Overall preflight result.')
     sheet_name: str = Field(description='Configured worksheet name used for preflight.')
     sheet_exists: bool = Field(description='Whether the configured worksheet was found.')
-    has_merged_header: bool | None = Field(
-        default=None,
-        description='Whether the workbook uses a merged two-row header when the header block was readable.',
-    )
+    has_merged_header: bool | None = Field(default=None, description='Whether the workbook uses a merged two-row header when the header block was readable.')
     estimated_row_count: int = Field(default=0, description='Estimated number of data rows for a later import run.')
-    structural_issue_codes: list[str] = Field(
-        default_factory=list,
-        description='Stable structural issue codes emitted for non-header preflight failures.',
-    )
+    structural_issue_codes: list[str] = Field(default_factory=list, description='Stable structural issue codes emitted for non-header preflight failures.')
 
     is_required_missing: bool = Field(default=False, description='Whether required headers are missing.')
-    missing_required: list[Label] = Field(
-        default_factory=empty_labels, description='Required headers missing from the workbook.'
-    )
-    missing_primary: list[Label] = Field(
-        default_factory=empty_labels, description='Primary-key headers missing from the workbook.'
-    )
-    unrecognized: list[Label] = Field(
-        default_factory=empty_labels, description='Headers present in the workbook but unknown to the schema.'
-    )
-    duplicated: list[Label] = Field(
-        default_factory=empty_labels, description='Headers that appear more than once in the workbook.'
-    )
+    missing_required: list[Label] = Field(default_factory=empty_labels, description='Required headers missing from the workbook.')
+    missing_primary: list[Label] = Field(default_factory=empty_labels, description='Primary-key headers missing from the workbook.')
+    unrecognized: list[Label] = Field(default_factory=empty_labels, description='Headers present in the workbook but unknown to the schema.')
+    duplicated: list[Label] = Field(default_factory=empty_labels, description='Headers that appear more than once in the workbook.')
+    # fmt: on
 
     @property
     def is_valid(self) -> bool:

src/excelalchemy/worksheet/header.py

@@ -10,11 +10,11 @@
 class ExcelHeader(BaseModel):
     """Normalized worksheet header extracted from user input."""
 
+    # fmt: off
     label: Label = Field(description='Worksheet header label.')
-    parent_label: Label = Field(
-        description='Parent worksheet header label. Falls back to the label itself for flat headers.'
-    )
+    parent_label: Label = Field(description='Parent worksheet header label. Falls back to the label itself for flat headers.')
     offset: int = Field(default=0, description='Child-column offset under a merged parent header.')
+    # fmt: on
 
     @property
     def unique_label(self) -> UniqueLabel:

tests/contracts/test_pydantic_contract.py

@@ -12,6 +12,9 @@
     ExcelFieldCodec,
     ExcelRowError,
     Label,
+    MultiChoiceCodec,
+    Option,
+    OptionId,
     ProgrammaticError,
 )
 from excelalchemy.adapters.pydantic import extract_pydantic_model, instantiate_pydantic_model
@@ -132,6 +135,49 @@ def must_use_company_domain(cls, value: str) -> str:
         assert isinstance(wrong_domain, list)
         assert wrong_domain == [ExcelCellError(label=Label('邮箱'), message='Must use the company domain')]
 
+    def test_instantiate_pydantic_model_validates_by_model_field_name_when_field_has_alias(self):
+        class AliasedModel(BaseModel):
+            full_name: Annotated[str, Field(alias='fullName'), ExcelColumn(label='姓名', order=1)]
+
+        result = instantiate_pydantic_model({'full_name': 'Alice'}, AliasedModel)
+
+        assert isinstance(result, AliasedModel)
+        assert result.full_name == 'Alice'
+
+    def test_instantiate_pydantic_model_maps_aliased_field_errors_to_excel_labels(self):
+        class AliasedModel(BaseModel):
+            full_name: Annotated[str, Field(alias='fullName', min_length=5), ExcelColumn(label='姓名', order=1)]
+
+        result = instantiate_pydantic_model({'full_name': 'Al'}, AliasedModel)
+
+        assert isinstance(result, list)
+        assert result == [
+            ExcelCellError(label=Label('姓名'), message='The minimum length is 5 characters', min_length=5)
+        ]
+
+    def test_instantiate_pydantic_model_normalizes_list_constraint_errors_from_pydantic_context(self):
+        options = [
+            Option(id=OptionId('a'), name='A'),
+            Option(id=OptionId('b'), name='B'),
+            Option(id=OptionId('c'), name='C'),
+        ]
+
+        class MultiChoiceModel(BaseModel):
+            choices: Annotated[
+                list[str],
+                Field(min_length=2, max_length=2),
+                ExcelColumn(codec=MultiChoiceCodec(), label='选项', order=1, options=options),
+            ]
+
+        too_short = instantiate_pydantic_model({'choices': ['A']}, MultiChoiceModel)
+        too_long = instantiate_pydantic_model({'choices': ['A', 'B', 'C']}, MultiChoiceModel)
+
+        assert isinstance(too_short, list)
+        assert too_short == [ExcelCellError(label=Label('选项'), message='Select at least 2 items', min_items=2)]
+
+        assert isinstance(too_long, list)
+        assert too_long == [ExcelCellError(label=Label('选项'), message='Select no more than 2 items', max_items=2)]
+
     def test_instantiate_pydantic_model_maps_model_validators_to_row_errors(self):
         class ModelValidatedContract(BaseModel):
             email: Annotated[str, ExcelColumn(codec=EmailCodec(), label='邮箱', order=1)]

tests/contracts/test_result_contract.py

@@ -1,4 +1,5 @@
 import pytest
+from pydantic import ValidationError
 
 from excelalchemy import ImportPreflightStatus, Label, ProgrammaticError, ValidateResult
 from excelalchemy.results import (
@@ -57,6 +58,10 @@ def test_import_result_returns_success_defaults_for_success_case(self):
         assert result.is_header_invalid is False
         assert result.is_data_invalid is False
 
+    def test_import_result_rejects_unknown_fields(self):
+        with pytest.raises(ValidationError):
+            ImportResult(result=ValidateResult.SUCCESS, legacy_field='kept')
+
     def test_import_result_to_api_payload_for_success_case(self):
         result = ImportResult(result=ValidateResult.SUCCESS, success_count=1, fail_count=0, url='memory://result.xlsx')