import datetime as _datetime
from typing import Any, Callable, Optional, Type, TypeVar, Union, overload, Literal
from . import (
DecodeError as _DecodeError,
convert as _convert,
to_builtins as _to_builtins,
)
__all__ = ("encode", "decode")
def __dir__():
return __all__
def _import_pyyaml(name):
try:
import yaml # type: ignore
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
`_ is installed.
See Also
--------
decode
"""
yaml = _import_pyyaml("encode")
# Use the C extension if available
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[bytes, 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
`_ is installed.
See Also
--------
encode
"""
yaml = _import_pyyaml("decode")
# Use the C extension if available
Loader = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
if not isinstance(buf, (str, bytes)):
# call `memoryview` first, since `bytes(1)` is actually valid
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,
)