mirror of
https://github.com/kennethreitz/pydantic.git
synced 2026-06-05 23:00:18 +00:00
Eric Jolibois
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
import datetime
|
|
from collections import deque
|
|
from decimal import Decimal
|
|
from enum import Enum
|
|
from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network
|
|
from pathlib import Path
|
|
from types import GeneratorType
|
|
from typing import Any, Callable, Dict, Pattern, Type, Union
|
|
from uuid import UUID
|
|
|
|
from .color import Color
|
|
from .types import SecretBytes, SecretStr
|
|
|
|
__all__ = 'pydantic_encoder', 'custom_pydantic_encoder', 'timedelta_isoformat'
|
|
|
|
|
|
def isoformat(o: Union[datetime.date, datetime.time]) -> str:
|
|
return o.isoformat()
|
|
|
|
|
|
ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
|
|
bytes: lambda o: o.decode(),
|
|
Color: str,
|
|
datetime.date: isoformat,
|
|
datetime.datetime: isoformat,
|
|
datetime.time: isoformat,
|
|
datetime.timedelta: lambda td: td.total_seconds(),
|
|
Decimal: float,
|
|
Enum: lambda o: o.value,
|
|
frozenset: list,
|
|
deque: list,
|
|
GeneratorType: list,
|
|
IPv4Address: str,
|
|
IPv4Interface: str,
|
|
IPv4Network: str,
|
|
IPv6Address: str,
|
|
IPv6Interface: str,
|
|
IPv6Network: str,
|
|
Path: str,
|
|
SecretBytes: str,
|
|
SecretStr: str,
|
|
set: list,
|
|
UUID: str,
|
|
}
|
|
|
|
|
|
def pydantic_encoder(obj: Any) -> Any:
|
|
from dataclasses import asdict, is_dataclass
|
|
|
|
from .main import BaseModel
|
|
|
|
if isinstance(obj, BaseModel):
|
|
return obj.dict()
|
|
elif is_dataclass(obj):
|
|
return asdict(obj)
|
|
|
|
if isinstance(obj, Pattern):
|
|
return obj.pattern
|
|
|
|
# Check the class type and its superclasses for a matching encoder
|
|
for base in obj.__class__.__mro__[:-1]:
|
|
try:
|
|
encoder = ENCODERS_BY_TYPE[base]
|
|
except KeyError:
|
|
continue
|
|
return encoder(obj)
|
|
else: # We have exited the for loop without finding a suitable encoder
|
|
raise TypeError(f"Object of type '{obj.__class__.__name__}' is not JSON serializable")
|
|
|
|
|
|
def custom_pydantic_encoder(type_encoders: Dict[Any, Callable[[Type[Any]], Any]], obj: Any) -> Any:
|
|
# Check the class type and its superclasses for a matching encoder
|
|
for base in obj.__class__.__mro__[:-1]:
|
|
try:
|
|
encoder = type_encoders[base]
|
|
except KeyError:
|
|
continue
|
|
return encoder(obj)
|
|
else: # We have exited the for loop without finding a suitable encoder
|
|
return pydantic_encoder(obj)
|
|
|
|
|
|
def timedelta_isoformat(td: datetime.timedelta) -> str:
|
|
"""
|
|
ISO 8601 encoding for timedeltas.
|
|
"""
|
|
minutes, seconds = divmod(td.seconds, 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
return f'P{td.days}DT{hours:d}H{minutes:d}M{seconds:d}.{td.microseconds:06d}S'
|