11414108386

Committed 19 Oct 2024 03:15AM UTC coverage: 98.447% (+0.1%) from 98.351%

Build # 11414108386

Build Type

Pull #35

github

Committed by

web-flow

Commit Message

Merge 2ea861c9a into 66ab444ec

Pull Request Pull Request #35: Serialization direct from proxy

Run Details

40 of 42 new or added lines in 6 files covered. (95.24%)

2 existing lines in 1 file now uncovered.

1521 of 1545 relevant lines covered (98.45%)

9.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.72

/src/numpydantic/interface/numpy.py

"""
Interface to numpy arrays
"""

from typing import Any, Literal, Optional, Union

from pydantic import BaseModel, SerializationInfo

from numpydantic.interface.interface import Interface, JsonDict
from numpydantic.serialization import pydantic_serializer

try:
    import numpy as np
    from numpy import ndarray

    ENABLED = True

except ImportError:  # pragma: no cover
    ENABLED = False
    ndarray = None
    np = None


class NumpyJsonDict(JsonDict):
    """
    JSON-able roundtrip representation of numpy array
    """

    type: Literal["numpy"]
    dtype: str
    value: list

    def to_array_input(self) -> ndarray:
        """
        Construct a numpy array
        """
        return np.array(self.value, dtype=self.dtype)


class SerializableNDArray(np.ndarray):
    """
    Trivial subclass of :class:`numpy.ndarray` that allows
    an additional ``__pydantic_serializer__`` attr to allow it to be
    json roundtripped without the help of an interface

    References:
        https://numpy.org/doc/stable/user/basics.subclassing.html#simple-example-adding-an-extra-attribute-to-ndarray
    """

    def __new__(cls, input_array: np.ndarray, **kwargs: dict[str, Any]):
        """Create a new ndarray instance, adding a new attribute"""
        obj = np.asarray(input_array, **kwargs).view(cls)
        obj.__pydantic_serializer__ = pydantic_serializer
        return obj

    def __array_finalize__(self, obj: Optional[np.ndarray]) -> None:
        if obj is None:
            return
        self.__pydantic_serializer__ = getattr(obj, "__pydantic_serializer__", None)


class NumpyInterface(Interface):
    """
    Numpy :class:`~numpy.ndarray` s!
    """

    name = "numpy"
    input_types = (ndarray, list)
    return_type = ndarray
    json_model = NumpyJsonDict
    priority = -999
    """
    The numpy interface is usually the interface of last resort.
    We want to use any more specific interface that we might have,
    because the numpy interface checks for anything that could be coerced
    to a numpy array (see :meth:`.NumpyInterface.check` )
    """

    @classmethod
    def check(cls, array: Any) -> bool:
        """
        Check that this is in fact a numpy ndarray or something that can be
        coerced to one
        """
        if array is None:
            return False

        if isinstance(array, (ndarray, SerializableNDArray)):
            return True
        elif isinstance(array, dict):
            return NumpyJsonDict.is_valid(array)
        else:
            try:
                _ = np.array(array)
                return True
            except Exception:
                return False

    def before_validation(self, array: Any) -> ndarray:
        """
        Coerce to an ndarray. We have already checked if coercion is possible
        in :meth:`.check`
        """
        if not isinstance(array, SerializableNDArray):
            array = SerializableNDArray(array)

        try:
            if issubclass(self.dtype, BaseModel) and isinstance(array.flat[0], dict):
                array = SerializableNDArray(
                    np.vectorize(lambda x: self.dtype(**x))(array)
                )
        except TypeError:
            # fine, dtype isn't a type
            pass

        return array

    @classmethod
    def enabled(cls) -> bool:
        """Check that numpy is present in the environment"""
        return ENABLED

    @classmethod
    def to_json(
        cls, array: ndarray, info: SerializationInfo = None
    ) -> Union[list, JsonDict]:
        """
        Convert an array of :attr:`.return_type` to a JSON-compatible format using
        base python types
        """
        if not isinstance(array, np.ndarray):  # pragma: no cover
            array = np.array(array)

        json_array = array.tolist()

        if info.round_trip:
            json_array = NumpyJsonDict(
                type=cls.name, dtype=str(array.dtype), value=json_array
            )
        return json_array

1	"""
2	Interface to numpy arrays
3	"""
4
5	from typing import Any, Literal, Optional, Union	10✔
6
7	from pydantic import BaseModel, SerializationInfo	10✔
8
9	from numpydantic.interface.interface import Interface, JsonDict	10✔
10	from numpydantic.serialization import pydantic_serializer	10✔
11
12	try:	10✔
13	import numpy as np	10✔
14	from numpy import ndarray	10✔
15
16	ENABLED = True	10✔
17
18	except ImportError: # pragma: no cover
19	ENABLED = False
20	ndarray = None
21	np = None
22
23
24	class NumpyJsonDict(JsonDict):	10✔
25	"""
26	JSON-able roundtrip representation of numpy array
27	"""
28
29	type: Literal["numpy"]	10✔
30	dtype: str	10✔
31	value: list	10✔
32
33	def to_array_input(self) -> ndarray:	10✔
34	"""
35	Construct a numpy array
36	"""
37	return np.array(self.value, dtype=self.dtype)	10✔
38
39
40	class SerializableNDArray(np.ndarray):	10✔
41	"""
42	Trivial subclass of :class:`numpy.ndarray` that allows
43	an additional ``__pydantic_serializer__`` attr to allow it to be
44	json roundtripped without the help of an interface
45
46	References:
47	https://numpy.org/doc/stable/user/basics.subclassing.html#simple-example-adding-an-extra-attribute-to-ndarray
48	"""
49
50	def __new__(cls, input_array: np.ndarray, **kwargs: dict[str, Any]):	10✔
51	"""Create a new ndarray instance, adding a new attribute"""
52	obj = np.asarray(input_array, **kwargs).view(cls)	10✔
53	obj.__pydantic_serializer__ = pydantic_serializer	10✔
54	return obj	10✔
55
56	def __array_finalize__(self, obj: Optional[np.ndarray]) -> None:	10✔
57	if obj is None:	10✔
NEW 58	return	×
59	self.__pydantic_serializer__ = getattr(obj, "__pydantic_serializer__", None)	10✔
60
61
62	class NumpyInterface(Interface):	10✔
63	"""
64	Numpy :class:`~numpy.ndarray` s!
65	"""
66
67	name = "numpy"	10✔
68	input_types = (ndarray, list)	10✔
69	return_type = ndarray	10✔
70	json_model = NumpyJsonDict	10✔
71	priority = -999	10✔
72	"""	6✔
73	The numpy interface is usually the interface of last resort.
74	We want to use any more specific interface that we might have,
75	because the numpy interface checks for anything that could be coerced
76	to a numpy array (see :meth:`.NumpyInterface.check` )
77	"""
78
79	@classmethod	10✔
80	def check(cls, array: Any) -> bool:	10✔
81	"""
82	Check that this is in fact a numpy ndarray or something that can be
83	coerced to one
84	"""
85	if array is None:	10✔
86	return False	×
87
88	if isinstance(array, (ndarray, SerializableNDArray)):	10✔
89	return True	10✔
90	elif isinstance(array, dict):	10✔
91	return NumpyJsonDict.is_valid(array)	10✔
92	else:
93	try:	10✔
94	_ = np.array(array)	10✔
95	return True	10✔
96	except Exception:	10✔
97	return False	10✔
98
99	def before_validation(self, array: Any) -> ndarray:	10✔
100	"""
101	Coerce to an ndarray. We have already checked if coercion is possible
102	in :meth:`.check`
103	"""
104	if not isinstance(array, SerializableNDArray):	10✔
105	array = SerializableNDArray(array)	10✔
106
107	try:	10✔
108	if issubclass(self.dtype, BaseModel) and isinstance(array.flat[0], dict):	10✔
109	array = SerializableNDArray(	10✔
110	np.vectorize(lambda x: self.dtype(**x))(array)
111	)
112	except TypeError:	10✔
113	# fine, dtype isn't a type
114	pass	10✔
115
116	return array	10✔
117
118	@classmethod	10✔
119	def enabled(cls) -> bool:	10✔
120	"""Check that numpy is present in the environment"""
121	return ENABLED	10✔
122
123	@classmethod	10✔
124	def to_json(	10✔
125	cls, array: ndarray, info: SerializationInfo = None
126	) -> Union[list, JsonDict]:
127	"""
128	Convert an array of :attr:`.return_type` to a JSON-compatible format using
129	base python types
130	"""
131	if not isinstance(array, np.ndarray): # pragma: no cover
132	array = np.array(array)
133
134	json_array = array.tolist()	10✔
135
136	if info.round_trip:	10✔
137	json_array = NumpyJsonDict(	10✔
138	type=cls.name, dtype=str(array.dtype), value=json_array
139	)
140	return json_array	10✔

p2p-ld / numpydantic / 11414108386

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous