E575 Move core schema generation logic for path types inside the `Generate… · pydantic/pydantic@aa85c3a · GitHub
[go: up one dir, main page]
Skip to content

Commit aa85c3a

Browse files
sydney-runkleViicos
sydney-runkle
and
Viicos
authored
Move core schema generation logic for path types inside the GenerateSchema class (#10846)
Co-authored-by: Victorien <65306057+Viicos@users.noreply.github.com>
1 parent aad4c43 commit aa85c3a

File tree

3 files changed

+53
-122
lines changed

3 files changed

+53
-122
lines changed

‎pydantic/_internal/_generate_schema.py‎

Lines changed: 52 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -480,6 +480,53 @@ def ser_ip(ip: Any, info: core_schema.SerializationInfo) -> str | IpType:
480480
},
481481
)
482482

483+
def _path_schema(self, tp: Any, path_type: Any) -> CoreSchema:
484+
if tp is os.PathLike and (path_type not in {str, bytes} and not _typing_extra.is_any(path_type)):
485+
raise PydanticUserError(
486+
'`os.PathLike` can only be used with `str`, `bytes` or `Any`', code='schema-for-unknown-type'
487+
)
488+
489+
path_constructor = pathlib.PurePath if tp is os.PathLike else tp
490+
constrained_schema = core_schema.bytes_schema() if (path_type is bytes) else core_schema.str_schema()
491+
492+
def path_validator(input_value: str | bytes) -> os.PathLike[Any]: # type: ignore
493+
try:
494+
if path_type is bytes:
495+
if isinstance(input_value, bytes):
496+
try:
497+
input_value = input_value.decode()
498+
except UnicodeDecodeError as e:
499+
raise PydanticCustomError('bytes_type', 'Input must be valid bytes') from e
500+
else:
501+
raise PydanticCustomError('bytes_type', 'Input must be bytes')
502+
elif not isinstance(input_value, str):
503+
raise PydanticCustomError('path_type', 'Input is not a valid path')
504+
505+
return path_constructor(input_value) # type: ignore
506+
except TypeError as e:
507+
raise PydanticCustomError('path_type', 'Input is not a valid path') from e
508+
509+
instance_schema = core_schema.json_or_python_schema(
510+
json_schema=core_schema.no_info_after_validator_function(path_validator, constrained_schema),
511+
python_schema=core_schema.is_instance_schema(tp),
512+
)
513+
514+
schema = core_schema.lax_or_strict_schema(
515+
lax_schema=core_schema.union_schema(
516+
[
517+
instance_schema,
518+
core_schema.no_info_after_validator_function(path_validator, constrained_schema),
519+
],
520+
custom_error_type='path_type',
521+
custom_error_message=f'Input is not a valid path for {tp}',
522+
strict=True,
523+
),
524+
strict_schema=instance_schema,
525+
serialization=core_schema.to_string_ser_schema(),
526+
metadata={'pydantic_js_functions': [lambda source, handler: {**handler(source), 'format': 'path'}]},
527+
)
528+
return schema
529+
483530
def _fraction_schema(self) -> CoreSchema:
484531
"""Support for [`fractions.Fraction`][fractions.Fraction]."""
485532
from ._validators import fraction_validator
@@ -944,6 +991,8 @@ def match_type(self, obj: Any) -> core_schema.CoreSchema: # noqa: C901
944991
return self._sequence_schema(Any)
945992
elif obj in DICT_TYPES:
946993
return self._dict_schema(Any, Any)
994+
elif obj in PATH_TYPES:
995+
return self._path_schema(obj, Any)
947996
elif _typing_extra.is_type_alias_type(obj):
948997
return self._type_alias_type_schema(obj)
949998
elif obj is type:
@@ -1022,6 +1071,8 @@ def _match_generic_type(self, obj: Any, origin: Any) -> CoreSchema: # noqa: C90
10221071
return self._frozenset_schema(self._get_first_arg_or_any(obj))
10231072
elif origin in DICT_TYPES:
10241073
return self._dict_schema(*self._get_first_two_args_or_any(obj))
1074+
elif origin in PATH_TYPES:
1075+
return self._path_schema(origin, self._get_first_arg_or_any(obj))
10251076
elif is_typeddict(origin):
10261077
return self._typed_dict_schema(obj, origin)
10271078
elif origin in (typing.Type, type):
@@ -1999,7 +2050,6 @@ def _get_prepare_pydantic_annotations_for_known_type(
19992050
from ._std_types_schema import (
20002051
deque_schema_prepare_pydantic_annotations,
20012052
mapping_like_prepare_pydantic_annotations,
2002-
path_schema_prepare_pydantic_annotations,
20032053
)
20042054

20052055
# Check for hashability
@@ -2014,9 +2064,7 @@ def _get_prepare_pydantic_annotations_for_known_type(
20142064
# not always called from match_type, but sometimes from _apply_annotations
20152065
obj_origin = get_origin(obj) or obj
20162066

2017-
if obj_origin in PATH_TYPES:
2018-
return path_schema_prepare_pydantic_annotations(obj, annotations)
2019-
elif obj_origin in DEQUE_TYPES:
2067+
if obj_origin in DEQUE_TYPES:
20202068
return deque_schema_prepare_pydantic_annotations(obj, annotations)
20212069
elif obj_origin in MAPPING_TYPES:
20222070
return mapping_like_prepare_pydantic_annotations(obj, annotations)

‎pydantic/_internal/_std_types_schema.py‎

Lines changed: 1 addition & 109 deletions
Original file line numberDiff line numberDiff line change
@@ -11,28 +11,24 @@
1111
import collections
1212
import collections.abc
1313
import dataclasses
14-
import os
1514
import typing
1615
from functools import partial
1716
from typing import Any, Callable, Iterable, Tuple, TypeVar, cast
1817

1918
import typing_extensions
2019
from pydantic_core import (
2120
CoreSchema,
22-
PydanticCustomError,
2321
core_schema,
2422
)
2523
from typing_extensions import get_args, get_origin
2624

2725
from pydantic._internal._serializers import serialize_sequence_via_list
2826
from pydantic.errors import PydanticSchemaGenerationError
29-
from pydantic.types import Strict
3027

31-
from ..json_schema import JsonSchemaValue
3228
from . import _known_annotated_metadata, _typing_extra
3329
from ._import_utils import import_cached_field_info
3430
from ._internal_dataclass import slots_true
35-
from ._schema_generation_shared import GetCoreSchemaHandler, GetJsonSchemaHandler
31+
from ._schema_generation_shared import GetCoreSchemaHandler
3632

3733
FieldInfo = import_cached_field_info()
3834

@@ -42,110 +38,6 @@
4238
StdSchemaFunction = Callable[[GenerateSchema, type[Any]], core_schema.CoreSchema]
4339

4440

45-
@dataclasses.dataclass(**slots_true)
46-
class InnerSchemaValidator:
47-
"""Use a fixed CoreSchema, avoiding interference from outward annotations."""
48-
49-
core_schema: CoreSchema
50-
js_schema: JsonSchemaValue | None = None
51-
js_core_schema: CoreSchema | None = None
52-
js_schema_update: JsonSchemaValue | None = None
53-
54-
def __get_pydantic_json_schema__(self, _schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue:
55-
if self.js_schema is not None:
56-
return self.js_schema
57-
js_schema = handler(self.js_core_schema or self.core_schema)
58-
if self.js_schema_update is not None:
59-
js_schema.update(self.js_schema_update)
60-
return js_schema
61-
62-
def __get_pydantic_core_schema__(self, _source_type: Any, _handler: GetCoreSchemaHandler) -> CoreSchema:
63-
return self.core_schema
64-
65-
66-
def path_schema_prepare_pydantic_annotations(
67-
source_type: Any, annotations: Iterable[Any]
68-
) -> tuple[Any, list[Any]] | None:
69-
import pathlib
70-
71-
orig_source_type: Any = get_origin(source_type) or source_type
72-
if (
73-
(source_type_args := get_args(source_type))
74-
and orig_source_type is os.PathLike
75-
and source_type_args[0] not in {str, bytes, Any}
76-
):
77-
return None
78-
79-
if orig_source_type not in {
80-
os.PathLike,
81-
pathlib.Path,
82-
pathlib.PurePath,
83-
pathlib.PosixPath,
84-
pathlib.PurePosixPath,
85-
pathlib.PureWindowsPath,
86-
}:
87-
return None
88-
89-
metadata, remaining_annotations = _known_annotated_metadata.collect_known_metadata(annotations)
90-
_known_annotated_metadata.check_metadata(metadata, _known_annotated_metadata.STR_CONSTRAINTS, orig_source_type)
91-
92-
is_first_arg_byte = source_type_args and source_type_args[0] is bytes
93-
construct_path = pathlib.PurePath if orig_source_type is os.PathLike else orig_source_type
94-
constrained_schema = (
95-
core_schema.bytes_schema(**metadata) if is_first_arg_byte else core_schema.str_schema(**metadata)
96-
)
97-
98-
def path_validator(input_value: str | bytes) -> os.PathLike[Any]: # type: ignore
99-
try:
100-
if is_first_arg_byte:
101-
if isinstance(input_value, bytes):
102-
try:
103-
input_value = input_value.decode()
104-
except UnicodeDecodeError as e:
105-
raise PydanticCustomError('bytes_type', 'Input must be valid bytes') from e
106-
else:
107-
raise PydanticCustomError('bytes_type', 'Input must be bytes')
108-
elif not isinstance(input_value, str):
109-
raise PydanticCustomError('path_type', 'Input is not a valid path')
110-
111-
return construct_path(input_value)
112-
except TypeError as e:
113-
raise PydanticCustomError('path_type', 'Input is not a valid path') from e
114-
115-
instance_schema = core_schema.json_or_python_schema(
116-
json_schema=core_schema.no_info_after_validator_function(path_validator, constrained_schema),
117-
python_schema=core_schema.is_instance_schema(orig_source_type),
118-
)
119-
120-
strict: bool | None = None
121-
for annotation in annotations:
122-
if isinstance(annotation, Strict):
123-
strict = annotation.strict
124-
125-
schema = core_schema.lax_or_strict_schema(
126-
lax_schema=core_schema.union_schema(
127-
[
128-
instance_schema,
129-
core_schema.no_info_after_validator_function(path_validator, constrained_schema),
130-
],
131-
custom_error_type='path_type',
132-
custom_error_message=f'Input is not a valid path for {orig_source_type}',
133-
strict=True,
134-
),
135-
strict_schema=instance_schema,
136-
serialization=core_schema.to_string_ser_schema(),
137-
strict=strict,
138-
)
139-
140-
return (
141-
orig_source_type,
142-
[
143-
InnerSchemaValidator(schema, js_core_schema=constrained_schema, js_schema_update={'format': 'path'}),
144-
*remaining_annotations,
145-
],
146-
)
147-
148-
14941
def deque_validator(
15042
input_value: Any, handler: core_schema.ValidatorFunctionWrapHandler, maxlen: None | int
15143
) -> collections.deque[Any]:

‎tests/test_types.py‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3498,15 +3498,6 @@ class Model(BaseModel):
34983498
assert Model.model_validate_json(json.dumps({'foo': str(value)})).foo == result
34993499

35003500

3501-
def test_path_validation_constrained():
3502-
ta = TypeAdapter(Annotated[Path, Field(min_length=9, max_length=20)])
3503-
with pytest.raises(ValidationError):
3504-
ta.validate_python('/short')
3505-
with pytest.raises(ValidationError):
3506-
ta.validate_python('/' + 'long' * 100)
3507-
assert ta.validate_python('/just/right/enough') == Path('/just/right/enough')
3508-
3509-
35103501
def test_path_like():
35113502
class Model(BaseModel):
35123503
foo: os.PathLike

0 commit comments

Comments
 (0)
0