-
-
Notifications
You must be signed in to change notification settings - Fork 481
/
Copy pathfiles.py
29 lines (23 loc) · 953 Bytes
/
files.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from typing import Any, Callable, Dict
from django.core.files.uploadedfile import UploadedFile as DjangoUploadedFile
from pydantic_core import core_schema
__all__ = ["UploadedFile"]
class UploadedFile(DjangoUploadedFile):
@classmethod
def __get_pydantic_json_schema__(
cls, core_schema: Any, handler: Callable[..., Any]
) -> Dict:
# calling handler(core_schema) here raises an exception
json_schema: Dict[str, str] = {}
json_schema.update(type="string", format="binary")
return json_schema
@classmethod
def _validate(cls, v: Any, _: Any) -> Any:
if not isinstance(v, DjangoUploadedFile):
raise ValueError(f"Expected UploadFile, received: {type(v)}")
return v
@classmethod
def __get_pydantic_core_schema__(
cls, source: Any, handler: Callable[..., Any]
) -> Any:
return core_schema.with_info_plain_validator_function(cls._validate)