| from __future__ import annotations |
|
|
| import datetime as _datetime |
| from typing import TYPE_CHECKING, overload, TypeVar, Any |
|
|
| from . import ( |
| DecodeError as _DecodeError, |
| convert as _convert, |
| to_builtins as _to_builtins, |
| ) |
|
|
| if TYPE_CHECKING: |
| from typing import Callable, Optional, Type, Union, Literal |
| from typing_extensions import Buffer |
|
|
|
|
| __all__ = ("encode", "decode") |
|
|
|
|
| def __dir__(): |
| return __all__ |
|
|
|
|
| def _import_pyyaml(name): |
| try: |
| import yaml |
| except ImportError: |
| raise ImportError( |
| f"`msgspec.yaml.{name}` requires PyYAML be installed.\n\n" |
| "Please either `pip` or `conda` install it as follows:\n\n" |
| " $ python -m pip install pyyaml # using pip\n" |
| " $ conda install pyyaml # or using conda" |
| ) from None |
| else: |
| return yaml |
|
|
|
|
| def encode( |
| obj: Any, |
| *, |
| enc_hook: Optional[Callable[[Any], Any]] = None, |
| order: Literal[None, "deterministic", "sorted"] = None, |
| ) -> bytes: |
| """Serialize an object as YAML. |
| |
| Parameters |
| ---------- |
| obj : Any |
| The object to serialize. |
| enc_hook : callable, optional |
| A callable to call for objects that aren't supported msgspec types. |
| Takes the unsupported object and should return a supported object, or |
| raise a ``NotImplementedError`` if unsupported. |
| order : {None, 'deterministic', 'sorted'}, optional |
| The ordering to use when encoding unordered compound types. |
| |
| - ``None``: All objects are encoded in the most efficient manner |
| matching their in-memory representations. The default. |
| - `'deterministic'`: Unordered collections (sets, dicts) are sorted to |
| ensure a consistent output between runs. Useful when |
| comparison/hashing of the encoded binary output is necessary. |
| - `'sorted'`: Like `'deterministic'`, but *all* object-like types |
| (structs, dataclasses, ...) are also sorted by field name before |
| encoding. This is slower than `'deterministic'`, but may produce more |
| human-readable output. |
| |
| Returns |
| ------- |
| data : bytes |
| The serialized object. |
| |
| Notes |
| ----- |
| This function requires that the third-party `PyYAML library |
| <https://pyyaml.org/>`_ is installed. |
| |
| See Also |
| -------- |
| decode |
| """ |
| yaml = _import_pyyaml("encode") |
| |
| Dumper = getattr(yaml, "CSafeDumper", yaml.SafeDumper) |
|
|
| return yaml.dump_all( |
| [ |
| _to_builtins( |
| obj, |
| builtin_types=(_datetime.datetime, _datetime.date), |
| enc_hook=enc_hook, |
| order=order, |
| ) |
| ], |
| encoding="utf-8", |
| Dumper=Dumper, |
| allow_unicode=True, |
| sort_keys=False, |
| ) |
|
|
|
|
| T = TypeVar("T") |
|
|
|
|
| @overload |
| def decode( |
| buf: Union[Buffer, str], |
| *, |
| strict: bool = True, |
| dec_hook: Optional[Callable[[type, Any], Any]] = None, |
| ) -> Any: |
| pass |
|
|
|
|
| @overload |
| def decode( |
| buf: Union[bytes, str], |
| *, |
| type: Type[T] = ..., |
| strict: bool = True, |
| dec_hook: Optional[Callable[[type, Any], Any]] = None, |
| ) -> T: |
| pass |
|
|
|
|
| @overload |
| def decode( |
| buf: Union[bytes, str], |
| *, |
| type: Any = ..., |
| strict: bool = True, |
| dec_hook: Optional[Callable[[type, Any], Any]] = None, |
| ) -> Any: |
| pass |
|
|
|
|
| def decode(buf, *, type=Any, strict=True, dec_hook=None): |
| """Deserialize an object from YAML. |
| |
| Parameters |
| ---------- |
| buf : bytes-like or str |
| The message to decode. |
| type : type, optional |
| A Python type (in type annotation form) to decode the object as. If |
| provided, the message will be type checked and decoded as the specified |
| type. Defaults to `Any`, in which case the message will be decoded |
| using the default YAML types. |
| strict : bool, optional |
| Whether type coercion rules should be strict. Setting to False enables |
| a wider set of coercion rules from string to non-string types for all |
| values. Default is True. |
| dec_hook : callable, optional |
| An optional callback for handling decoding custom types. Should have |
| the signature ``dec_hook(type: Type, obj: Any) -> Any``, where ``type`` |
| is the expected message type, and ``obj`` is the decoded representation |
| composed of only basic YAML types. This hook should transform ``obj`` |
| into type ``type``, or raise a ``NotImplementedError`` if unsupported. |
| |
| Returns |
| ------- |
| obj : Any |
| The deserialized object. |
| |
| Notes |
| ----- |
| This function requires that the third-party `PyYAML library |
| <https://pyyaml.org/>`_ is installed. |
| |
| See Also |
| -------- |
| encode |
| """ |
| yaml = _import_pyyaml("decode") |
| |
| Loader = getattr(yaml, "CSafeLoader", yaml.SafeLoader) |
| if not isinstance(buf, (str, bytes)): |
| |
| buf = bytes(memoryview(buf)) |
| try: |
| obj = yaml.load(buf, Loader) |
| except yaml.YAMLError as exc: |
| raise _DecodeError(str(exc)) from None |
|
|
| if type is Any: |
| return obj |
| return _convert( |
| obj, |
| type, |
| builtin_types=(_datetime.datetime, _datetime.date), |
| strict=strict, |
| dec_hook=dec_hook, |
| ) |
|
|