Coverage for src/csv_schema_validator/core/models.py: 100%

47 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-23 15:34 +0100

1""" 

2Pydantic models for CSV schema definition. 

3""" 

4from __future__ import annotations 

5 

6from enum import Enum 

7from typing import ClassVar 

8 

9from pydantic import BaseModel, Field, field_validator, model_validator 

10 

11 

12class FieldType(str, Enum): 

13 STRING = "string" 

14 NUMBER = "number" 

15 INTEGER = "integer" 

16 BOOLEAN = "boolean" 

17 

18 

19class FieldSchema(BaseModel): 

20 """Schema definition for a single CSV field.""" 

21 

22 name: str = Field(..., description="Field name") 

23 type: FieldType = Field(..., description="Field data type") 

24 required: bool = Field(..., description="Whether the field is required") 

25 description: str | None = Field(None, description="Field description") 

26 pattern: str | None = Field(None, description="Field pattern") 

27 enum: list[str] | None = Field(None, description="Field enum") 

28 min: int | None = Field(None, description="Field minimum value") 

29 max: int | None = Field(None, description="Field maximum value") 

30 

31 # Class variables for validation 

32 _SUPPORTED_TYPES: ClassVar[set[str]] = {"string", "number", "integer", "boolean"} 

33 _NUMERIC_TYPES: ClassVar[set[str]] = {"number", "integer"} 

34 

35 @field_validator("pattern") 

36 @classmethod 

37 def validate_pattern(cls, v: str | None) -> str | None: 

38 """Validate that the pattern is a valid regex.""" 

39 if v is not None: 

40 import re 

41 try: 

42 re.compile(v) 

43 except re.error as e: 

44 raise ValueError(f"Invalid pattern: {e}") 

45 return v 

46 

47 @model_validator(mode="after") 

48 def validate_min_max(self) -> FieldSchema: 

49 """Validate that min/max constraints are only applied to numeric types.""" 

50 if self.min is not None or self.max is not None: 

51 if self.type.value not in self._NUMERIC_TYPES: 

52 raise ValueError( 

53 f"min/max constraints not applicable for {self.type.value} type" 

54 ) 

55 return self 

56 

57 

58class CSVSchema(BaseModel): 

59 """Schema definition for a CSV file.""" 

60 

61 name: str = Field(..., description="Schema name") 

62 description: str | None = Field(None, description="Schema description") 

63 fields: list[FieldSchema] = Field(..., min_length=1, description="Field definitions") 

64 

65 @field_validator("fields") 

66 @classmethod 

67 def validate_fields(cls, v: list[FieldSchema]) -> list[FieldSchema]: 

68 """Validate that field names are unique.""" 

69 names = [field.name for field in v] 

70 if len(names) != len(set(names)): 

71 raise ValueError("Field names must be unique") 

72 return v