26861324571

Committed 03 Jun 2026 03:09AM UTC coverage: 96.957% (-0.9%) from 97.821%

Build # 26861324571

Build Type

Pull #69

github

Committed by

web-flow

Commit Message

Merge 52b2ad15f into 952a740e0

Pull Request Pull Request #69: aw shit it's mypy plugin time

Coverage Stats

370 of 400 new or added lines in 14 files covered. (92.5%)

4 existing lines in 1 file now uncovered.

1912 of 1972 relevant lines covered (96.96%)

4.84 hits per line

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

96.94

/src/numpydantic/interface/interface.py

"""
Base Interface metaclass
"""

import inspect
import warnings
from abc import ABC, abstractmethod
from functools import lru_cache
from importlib.metadata import PackageNotFoundError, version
from operator import attrgetter
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, Union

if TYPE_CHECKING:
    from numpydantic.interface.typing import InterfaceTyping

import numpy as np
from pydantic import BaseModel, SerializationInfo, ValidationError

from numpydantic.exceptions import (
    DtypeError,
    MarkMismatchError,
    NoMatchError,
    ShapeError,
    TooManyMatchesError,
)
from numpydantic.types import DtypeType, NDArrayType, ShapeType
from numpydantic.validation import validate_dtype, validate_shape

T = TypeVar("T", bound=NDArrayType)
U = TypeVar("U", bound="JsonDict")
V = TypeVar("V")  # input type
W = TypeVar("W")  # Any type in handle_input


class InterfaceMark(BaseModel):
    """JSON-able mark to be able to round-trip json dumps"""

    module: str
    cls: str
    name: str
    version: str

    def is_valid(self, cls: type["Interface"], raise_on_error: bool = False) -> bool:
        """
        Check that a given interface matches the mark.

        Args:
            cls (Type): Interface type to check
            raise_on_error (bool): Raise an ``MarkMismatchError`` when the match
                is incorrect

        Returns:
            bool

        Raises:
            :class:`.MarkMismatchError` if requested by ``raise_on_error``
            for an invalid match
        """
        mark = cls.mark_interface()
        valid = self == mark
        if not valid and raise_on_error:
            raise MarkMismatchError(
                "Mismatch between serialized mark and current interface, "
                f"Serialized: {self}; current: {cls}"
            )
        return valid

    def match_by_name(self) -> type["Interface"] | None:
        """
        Try to find a matching interface by its name, returning it if found,
        or None if not found.
        """
        for i in Interface.interfaces(sort=False):
            if i.name == self.name:
                return i
        return None


class JsonDict(BaseModel):
    """
    Representation of array when dumped with round_trip == True.

    .. admonition:: Developer's Note

        Any JsonDict that contains an actual array should be named ``value``
        rather than array (or any other name), and nothing but the
        array data should be named ``value`` .

        During JSON serialization, it becomes ambiguous what contains an array
        of data vs. an array of metadata. For the moment we would like to
        reserve the ability to have lists of metadata, so until we rule that out,
        we would like to be able to avoid iterating over every element of an array
        in any context parameter transformation like relativizing/absolutizing paths.
        To avoid that, it's good to agree on a single value name -- ``value`` --
        and avoid using it for anything else.

    """

    type: str

    @abstractmethod
    def to_array_input(self) -> V:
        """
        Convert this roundtrip specifier to the relevant input class
        (one of the ``input_types`` of an interface).
        """

    @classmethod
    def is_valid(cls, val: dict, raise_on_error: bool = False) -> bool:
        """
        Check whether a given dictionary matches this JsonDict specification

        Args:
            val (dict): The dictionary to check for validity
            raise_on_error (bool): If ``True``, raise the validation error
                rather than returning a bool. (default: ``False``)

        Returns:
            bool - true if valid, false if not
        """
        try:
            _ = cls.model_validate(val)
            return True
        except ValidationError as e:
            if raise_on_error:
                raise e
            return False

    @classmethod
    def handle_input(cls: type[U], value: dict | U | W) -> V | W:
        """
        Handle input that is the json serialized roundtrip version
        (from :func:`~pydantic.BaseModel.model_dump` with ``round_trip=True``)
        converting it to the input format with :meth:`.JsonDict.to_array_input`
        or passing it through if not applicable
        """
        if isinstance(value, dict):
            value = cls(**value).to_array_input()
        elif isinstance(value, cls):
            value = value.to_array_input()
        return value

    @staticmethod
    def reshape_input(value: T, shape: tuple[int, ...]) -> T:
        """
        If a `reshape` value is present on the array, and the array shape doesn't match,
        attempt to reshape it.
        """
        if value.shape != shape:
            try:
                value = value.reshape(shape)
            except ValueError:
                warnings.warn(
                    f"Input data has shape {value.shape}, "
                    f"but roundtrip form specifies {shape},"
                    f"and {value.shape} can't be cast to {shape}. "
                    f"Attempting to proceed with validation without reshaping.",
                    stacklevel=1,
                )
        return value


class MarkedJson(BaseModel):
    """
    Model of JSON dumped with an additional interface mark
    with ``model_dump_json({'mark_interface': True})``
    """

    interface: InterfaceMark
    value: list | dict
    """
    Inner value of the array, we don't validate for JsonDict here, 
    that should be downstream from us for performance reasons 
    """

    @classmethod
    def try_cast(cls, value: V | dict) -> Union[V, "MarkedJson"]:
        """
        Try to cast to MarkedJson if applicable, otherwise return input
        """
        if isinstance(value, dict) and "interface" in value and "value" in value:
            try:
                value = MarkedJson(**value)
            except ValidationError:
                # fine, just not a MarkedJson dict even if it looks like one
                return value
        return value


class Interface(ABC, Generic[T]):
    """
    Abstract parent class for interfaces to different array formats
    """

    input_types: tuple[Any, ...]
    return_type: type[T]
    priority: int = 0
    typing: ClassVar[type["InterfaceTyping"] | None] = None
    """
    Optional static-typing companion class used by the mypy plugin and
    the mypy test generator. ``None`` means this interface does not opt
    into static constructor inference.
    """

    def __init__(self, shape: ShapeType = Any, dtype: DtypeType = Any) -> None:
        self.shape = shape
        self.dtype = dtype

    def validate(self, array: Any) -> T:
        """
        Validate input, returning final array type

        Calls the methods, in order:

        * array = :meth:`.deserialize` (array)
        * array = :meth:`.before_validation` (array)
        * dtype = :meth:`.get_dtype` (array) - get the dtype from the array,
            override if eg. the dtype is not contained in ``array.dtype``
        * valid = :meth:`.validate_dtype` (dtype) - check that the dtype matches
            the one in the NDArray specification. Override if special
            validation logic is needed for a given format
        * :meth:`.raise_for_dtype` (valid, dtype) - after checking dtype validity,
            raise an exception if it was invalid. Override to implement custom
            exceptions or error conditions, or make validation errors conditional.
        * array = :meth:`.after_validate_dtype` (array) - hook for additional
            validation or array modification mid-validation
        * shape = :meth:`.get_shape` (array) - get the shape from the array,
            override if eg. the shape is not contained in ``array.shape``
        * valid = :meth:`.validate_shape` (shape) - check that the shape matches
            the one in the NDArray specification. Override if special validation
            logic is needed.
        * :meth:`.raise_for_shape` (valid, shape) - after checking shape validity,
            raise an exception if it was invalid. You know the deal bc it's the same
            as raise for dtype.
        * :meth:`.after_validation` - hook after validation for modifying the array
            that is set as the model field value

        Follow the method signatures and return types to override.

        Implementing an interface subclass largely consists of overriding these methods
        as needed.

        Raises:
            If validation fails, rather than eg. returning ``False``, exceptions will
            be raised (to halt the rest of the pydantic validation process).
            When using interfaces outside of pydantic, you must catch both
            :class:`.DtypeError` and :class:`.ShapeError` (both of which are children
            of :class:`.InterfaceError` )
        """
        array = self.deserialize(array)

        array = self.before_validation(array)

        dtype = self.get_dtype(array)
        dtype_valid = self.validate_dtype(dtype)
        self.raise_for_dtype(dtype_valid, dtype)
        array = self.after_validate_dtype(array)

        shape = self.get_shape(array)
        shape_valid = self.validate_shape(shape)
        self.raise_for_shape(shape_valid, shape)

        array = self.after_validation(array)

        return array

    def deserialize(self, array: Any) -> V | Any:
        """
        If given a JSON serialized version of the array,
        deserialize it first.

        If a roundtrip-serialized :class:`.JsonDict`,
        pass to :meth:`.JsonDict.handle_input`.

        If a roundtrip-serialized :class:`.MarkedJson`,
        unpack mark, check for validity, warn if not,
        and try to continue with validation
        """
        if isinstance(marked_array := MarkedJson.try_cast(array), MarkedJson):
            try:
                marked_array.interface.is_valid(self.__class__, raise_on_error=True)
            except MarkMismatchError as e:
                warnings.warn(
                    str(e) + "\nAttempting to continue validation...", stacklevel=2
                )
            array = marked_array.value

        return self.json_model.handle_input(array)

    def before_validation(self, array: Any) -> NDArrayType:
        """
        Optional step pre-validation that coerces the input into a type that can be
        validated for shape and dtype

        Default method is a no-op
        """
        return array

    def get_dtype(self, array: NDArrayType) -> DtypeType:
        """
        Get the dtype from the input array.
        """
        if hasattr(array.dtype, "type") and array.dtype.type is np.object_:
            return self.get_object_dtype(array)
        else:
            return array.dtype

    def get_object_dtype(self, array: NDArrayType) -> DtypeType:
        """
        When an array contains an object, get the dtype of the object contained
        by the array.

        If this method returns `Any`, the dtype validation passes -
        used for e.g. empty arrays for which the dtype of the array can't be determined
        (since there are no objects).
        """
        try:
            return type(array.ravel()[0])
        except IndexError:
            return Any

    def validate_dtype(self, dtype: DtypeType) -> bool:
        """
        Validate the dtype of the given array, returning
        ``True`` if valid, ``False`` if not.
        """
        return validate_dtype(dtype, self.dtype)

    def raise_for_dtype(self, valid: bool, dtype: DtypeType) -> None:
        """
        After validating, raise an exception if invalid
        Raises:
            :class:`~numpydantic.exceptions.DtypeError`
        """
        if not valid:
            raise DtypeError(f"Invalid dtype! expected {self.dtype}, got {dtype}")

    def after_validate_dtype(self, array: NDArrayType) -> NDArrayType:
        """
        Hook to modify array after validating dtype.
        Default is a no-op.
        """
        return array

    def get_shape(self, array: NDArrayType) -> tuple[int, ...]:
        """
        Get the shape from the array as a tuple of integers
        """
        return array.shape

    def validate_shape(self, shape: tuple[int, ...]) -> bool:
        """
        Validate the shape of the given array against the shape
        specifier, returning ``True`` if valid, ``False`` if not.


        """
        if self.shape is Any:
            return True

        return validate_shape(shape, self.shape)

    def raise_for_shape(self, valid: bool, shape: tuple[int, ...]) -> None:
        """
        Raise a ShapeError if the shape is invalid.

        Raises:
            :class:`~numpydantic.exceptions.ShapeError`
        """
        if not valid:
            raise ShapeError(
                f"Invalid shape! expected shape {self.shape.prepared_args}, "
                f"got shape {shape}"
            )

    def after_validation(self, array: NDArrayType) -> T:
        """
        Optional step post-validation that coerces the intermediate array type into the
        return type

        Default method is a no-op
        """
        return array

    @classmethod
    @abstractmethod
    def check(cls, array: Any) -> bool:
        """
        Method to check whether a given input applies to this interface
        """

    @classmethod
    @abstractmethod
    def enabled(cls) -> bool:
        """
        Check whether this array interface can be used (eg. its dependent packages are
        installed, etc.)
        """

    @property
    @abstractmethod
    def name(self) -> str:
        """
        Short name for this interface
        """

    @property
    @abstractmethod
    def json_model(self) -> JsonDict:
        """
        The :class:`.JsonDict` model used for roundtripping
        JSON serialization
        """

    @classmethod
    @abstractmethod
    def to_json(cls, array: type[T], info: SerializationInfo) -> list | JsonDict:
        """
        Convert an array of :attr:`.return_type` to a JSON-compatible format using
        base python types
        """

    @classmethod
    def mark_json(cls, array: list | dict) -> dict:
        """
        When using ``model_dump_json`` with ``mark_interface: True`` in the ``context``,
        add additional annotations that would allow the serialized array to be
        roundtripped.

        Default is just to add an :class:`.InterfaceMark`

        Examples:

            >>> from pprint import pprint
            >>> pprint(Interface.mark_json([1.0, 2.0]))
            {'interface': {'cls': 'Interface',
                           'module': 'numpydantic.interface.interface',
                           'version': '1.2.2'},
             'value': [1.0, 2.0]}
        """
        return {"interface": cls.mark_interface(), "value": array}

    @classmethod
    def interfaces(
        cls, with_disabled: bool = False, sort: bool = True
    ) -> tuple[type["Interface"], ...]:
        """
        Enabled interface subclasses

        Args:
            with_disabled (bool): If ``True`` , get every known interface.
                If ``False`` (default), get only enabled interfaces.
            sort (bool): If ``True`` (default), sort interfaces by priority.
                If ``False`` , sorted by definition order. Used for recursion:
                we only want to sort once at the top level.
        """
        # get recursively
        subclasses = []
        for i in cls.__subclasses__():
            if with_disabled:
                subclasses.append(i)

            if i.enabled():
                subclasses.append(i)

            subclasses.extend(i.interfaces(with_disabled=with_disabled, sort=False))

        if sort:
            subclasses = sorted(
                subclasses,
                key=attrgetter("priority"),
                reverse=True,
            )

        return tuple(subclasses)

    @classmethod
    def return_types(cls) -> tuple[NDArrayType, ...]:
        """Return types for all enabled interfaces"""
        return tuple([i.return_type for i in cls.interfaces()])

    @classmethod
    def input_types(cls) -> tuple[Any, ...]:
        """Input types for all enabled interfaces"""
        in_types = []
        for iface in cls.interfaces():
            if isinstance(iface.input_types, (tuple, list)):
                in_types.extend(iface.input_types)
            else:  # pragma: no cover
                in_types.append(iface.input_types)

        return tuple(in_types)

    @classmethod
    def match_mark(cls, array: Any) -> type["Interface"] | None:
        """
        Match a marked JSON dump of this array to the interface that it indicates.

        First find an interface that matches by name, and then run its
        ``check`` method, because arrays can be dumped with a mark
        but without ``round_trip == True`` (and thus can't necessarily
        use the same interface that they were dumped with)

        Returns:
            Interface if match found, None otherwise
        """
        mark = MarkedJson.try_cast(array)
        if not isinstance(mark, MarkedJson):
            return None

        interface = mark.interface.match_by_name()
        if interface is not None and interface.check(mark.value):
            return interface
        return None

    @classmethod
    def match(cls, array: Any, fast: bool = False) -> type["Interface"]:
        """
        Find the interface that should be used for this array based on its input type

        First runs the ``check`` method for all interfaces returned by
        :meth:`.Interface.interfaces` **except** for :class:`.NumpyInterface` ,
        and if no match is found then try the numpy interface. This is because
        :meth:`.NumpyInterface.check` can be expensive, as we could potentially
        try to

        Args:
            fast (bool): if ``False`` , check all interfaces and raise exceptions for
              having multiple matching interfaces (default). If ``True`` ,
              check each interface (as ordered by its ``priority`` , decreasing),
              and return on the first match.
        """
        # Shortcircuit match if this is a marked json dump
        array = MarkedJson.try_cast(array)
        if (match := cls.match_mark(array)) is not None:
            return match
        elif isinstance(array, MarkedJson):
            array = array.value

        # first try and find a non-numpy interface, since the numpy interface
        # will try and load the array into memory in its check method
        interfaces = cls.interfaces()
        non_np_interfaces = [i for i in interfaces if i.name != "numpy"]
        np_interface = [i for i in interfaces if i.name == "numpy"][0]

        if fast:
            matches = []
            for i in non_np_interfaces:
                if i.check(array):
                    return i
        else:
            matches = [i for i in non_np_interfaces if i.check(array)]

        if len(matches) > 1:
            msg = f"More than one interface matches input {array}:\n"
            msg += "\n".join([f"  - {i}" for i in matches])
            raise TooManyMatchesError(msg)
        elif len(matches) == 0:
            # now try the numpy interface
            if np_interface.check(array):
                return np_interface
            else:
                raise NoMatchError(f"No matching interfaces found for input {array}")
        else:
            return matches[0]

    @classmethod
    def match_output(cls, array: Any) -> type["Interface"]:
        """
        Find the interface that should be used based on the output type -
        in the case that the output type differs from the input type, eg.
        the HDF5 interface, match an instantiated array for purposes of
        serialization to json, etc.
        """
        matches = [i for i in cls.interfaces() if isinstance(array, i.return_type)]
        if len(matches) > 1:
            msg = f"More than one interface matches output {array}:\n"
            msg += "\n".join([f"  - {i}" for i in matches])
            raise TooManyMatchesError(msg)
        elif len(matches) == 0:
            raise NoMatchError(f"No matching interfaces found for output {array}")
        else:
            return matches[0]

    @classmethod
    @lru_cache(maxsize=32)
    def mark_interface(cls) -> InterfaceMark:
        """
        Create an interface mark indicating this interface for validation after
        JSON serialization with ``round_trip==True``
        """
        interface_module = inspect.getmodule(cls)
        interface_module = (
            None if interface_module is None else interface_module.__name__
        )
        try:
            v = (
                None
                if interface_module is None
                else version(interface_module.split(".")[0])
            )
        except (
            PackageNotFoundError
        ):  # pragma: no cover - no tests for missing interface deps
            v = None

        return InterfaceMark(
            module=interface_module, cls=cls.__name__, name=cls.name, version=v
        )


class Proxy(ABC):
    """
    A proxy class that exposes some non-array data source (like a video) as an array
    """

    @classmethod
    @abstractmethod
    def proxy_for(cls) -> type[Interface]:
        """
        Declare the interface that this is a proxy for,
        allowing the proxy to be used with the NDArraySchema annotation
        with any of the input types that the Interface supports.
        """
        raise NotImplementedError()

1	"""
2	Base Interface metaclass
3	"""
4
5	import inspect	5✔
6	import warnings	5✔
7	from abc import ABC, abstractmethod	5✔
8	from functools import lru_cache	5✔
9	from importlib.metadata import PackageNotFoundError, version	5✔
10	from operator import attrgetter	5✔
11	from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, Union	5✔
12
13	if TYPE_CHECKING:	5✔
NEW 14	from numpydantic.interface.typing import InterfaceTyping	×
15
16	import numpy as np	5✔
17	from pydantic import BaseModel, SerializationInfo, ValidationError	5✔
18
19	from numpydantic.exceptions import (	5✔
20	DtypeError,
21	MarkMismatchError,
22	NoMatchError,
23	ShapeError,
24	TooManyMatchesError,
25	)
26	from numpydantic.types import DtypeType, NDArrayType, ShapeType	5✔
27	from numpydantic.validation import validate_dtype, validate_shape	5✔
28
29	T = TypeVar("T", bound=NDArrayType)	5✔
30	U = TypeVar("U", bound="JsonDict")	5✔
31	V = TypeVar("V") # input type	5✔
32	W = TypeVar("W") # Any type in handle_input	5✔
33
34
35	class InterfaceMark(BaseModel):	5✔
36	"""JSON-able mark to be able to round-trip json dumps"""
37
38	module: str	5✔
39	cls: str	5✔
40	name: str	5✔
41	version: str	5✔
42
43	def is_valid(self, cls: type["Interface"], raise_on_error: bool = False) -> bool:	5✔
44	"""
45	Check that a given interface matches the mark.
46
47	Args:
48	cls (Type): Interface type to check
49	raise_on_error (bool): Raise an ``MarkMismatchError`` when the match
50	is incorrect
51
52	Returns:
53	bool
54
55	Raises:
56	:class:`.MarkMismatchError` if requested by ``raise_on_error``
57	for an invalid match
58	"""
59	mark = cls.mark_interface()	5✔
60	valid = self == mark	5✔
61	if not valid and raise_on_error:	5✔
62	raise MarkMismatchError(	5✔
63	"Mismatch between serialized mark and current interface, "
64	f"Serialized: {self}; current: {cls}"
65	)
66	return valid	5✔
67
68	def match_by_name(self) -> type["Interface"] \| None:	5✔
69	"""
70	Try to find a matching interface by its name, returning it if found,
71	or None if not found.
72	"""
73	for i in Interface.interfaces(sort=False):	5✔
74	if i.name == self.name:	5✔
75	return i	5✔
76	return None	5✔
77
78
79	class JsonDict(BaseModel):	5✔
80	"""
81	Representation of array when dumped with round_trip == True.
82
83	.. admonition:: Developer's Note
84
85	Any JsonDict that contains an actual array should be named ``value``
86	rather than array (or any other name), and nothing but the
87	array data should be named ``value`` .
88
89	During JSON serialization, it becomes ambiguous what contains an array
90	of data vs. an array of metadata. For the moment we would like to
91	reserve the ability to have lists of metadata, so until we rule that out,
92	we would like to be able to avoid iterating over every element of an array
93	in any context parameter transformation like relativizing/absolutizing paths.
94	To avoid that, it's good to agree on a single value name -- ``value`` --
95	and avoid using it for anything else.
96
97	"""
98
99	type: str	5✔
100
101	@abstractmethod	5✔
102	def to_array_input(self) -> V:	5✔
103	"""
104	Convert this roundtrip specifier to the relevant input class
105	(one of the ``input_types`` of an interface).
106	"""
107
108	@classmethod	5✔
109	def is_valid(cls, val: dict, raise_on_error: bool = False) -> bool:	5✔
110	"""
111	Check whether a given dictionary matches this JsonDict specification
112
113	Args:
114	val (dict): The dictionary to check for validity
115	raise_on_error (bool): If ``True``, raise the validation error
116	rather than returning a bool. (default: ``False``)
117
118	Returns:
119	bool - true if valid, false if not
120	"""
121	try:	5✔
122	_ = cls.model_validate(val)	5✔
123	return True	5✔
124	except ValidationError as e:	5✔
125	if raise_on_error:	5✔
126	raise e	5✔
127	return False	5✔
128
129	@classmethod	5✔
130	def handle_input(cls: type[U], value: dict \| U \| W) -> V \| W:	5✔
131	"""
132	Handle input that is the json serialized roundtrip version
133	(from :func:`~pydantic.BaseModel.model_dump` with ``round_trip=True``)
134	converting it to the input format with :meth:`.JsonDict.to_array_input`
135	or passing it through if not applicable
136	"""
137	if isinstance(value, dict):	5✔
138	value = cls(**value).to_array_input()	5✔
139	elif isinstance(value, cls):	5✔
140	value = value.to_array_input()	5✔
141	return value	5✔
142
143	@staticmethod	5✔
144	def reshape_input(value: T, shape: tuple[int, ...]) -> T:	5✔
145	"""
146	If a `reshape` value is present on the array, and the array shape doesn't match,
147	attempt to reshape it.
148	"""
149	if value.shape != shape:	5✔
150	try:	5✔
151	value = value.reshape(shape)	5✔
152	except ValueError:	×
153	warnings.warn(	×
154	f"Input data has shape {value.shape}, "
155	f"but roundtrip form specifies {shape},"
156	f"and {value.shape} can't be cast to {shape}. "
157	f"Attempting to proceed with validation without reshaping.",
158	stacklevel=1,
159	)
160	return value	5✔
161
162
163	class MarkedJson(BaseModel):	5✔
164	"""
165	Model of JSON dumped with an additional interface mark
166	with ``model_dump_json({'mark_interface': True})``
167	"""
168
169	interface: InterfaceMark	5✔
170	value: list \| dict	5✔
171	"""	5✔
172	Inner value of the array, we don't validate for JsonDict here,
173	that should be downstream from us for performance reasons
174	"""
175
176	@classmethod	5✔
177	def try_cast(cls, value: V \| dict) -> Union[V, "MarkedJson"]:	5✔
178	"""
179	Try to cast to MarkedJson if applicable, otherwise return input
180	"""
181	if isinstance(value, dict) and "interface" in value and "value" in value:	5✔
182	try:	5✔
183	value = MarkedJson(**value)	5✔
184	except ValidationError:	5✔
185	# fine, just not a MarkedJson dict even if it looks like one
186	return value	5✔
187	return value	5✔
188
189
190	class Interface(ABC, Generic[T]):	5✔
191	"""
192	Abstract parent class for interfaces to different array formats
193	"""
194
195	input_types: tuple[Any, ...]	3✔
196	return_type: type[T]	3✔
197	priority: int = 0	5✔
198	typing: ClassVar[type["InterfaceTyping"] \| None] = None	5✔
199	"""	5✔
200	Optional static-typing companion class used by the mypy plugin and
201	the mypy test generator. ``None`` means this interface does not opt
202	into static constructor inference.
203	"""
204
205	def __init__(self, shape: ShapeType = Any, dtype: DtypeType = Any) -> None:	5✔
206	self.shape = shape	5✔
207	self.dtype = dtype	5✔
208
209	def validate(self, array: Any) -> T:	5✔
210	"""
211	Validate input, returning final array type
212
213	Calls the methods, in order:
214
215	* array = :meth:`.deserialize` (array)
216	* array = :meth:`.before_validation` (array)
217	* dtype = :meth:`.get_dtype` (array) - get the dtype from the array,
218	override if eg. the dtype is not contained in ``array.dtype``
219	* valid = :meth:`.validate_dtype` (dtype) - check that the dtype matches
220	the one in the NDArray specification. Override if special
221	validation logic is needed for a given format
222	* :meth:`.raise_for_dtype` (valid, dtype) - after checking dtype validity,
223	raise an exception if it was invalid. Override to implement custom
224	exceptions or error conditions, or make validation errors conditional.
225	* array = :meth:`.after_validate_dtype` (array) - hook for additional
226	validation or array modification mid-validation
227	* shape = :meth:`.get_shape` (array) - get the shape from the array,
228	override if eg. the shape is not contained in ``array.shape``
229	* valid = :meth:`.validate_shape` (shape) - check that the shape matches
230	the one in the NDArray specification. Override if special validation
231	logic is needed.
232	* :meth:`.raise_for_shape` (valid, shape) - after checking shape validity,
233	raise an exception if it was invalid. You know the deal bc it's the same
234	as raise for dtype.
235	* :meth:`.after_validation` - hook after validation for modifying the array
236	that is set as the model field value
237
238	Follow the method signatures and return types to override.
239
240	Implementing an interface subclass largely consists of overriding these methods
241	as needed.
242
243	Raises:
244	If validation fails, rather than eg. returning ``False``, exceptions will
245	be raised (to halt the rest of the pydantic validation process).
246	When using interfaces outside of pydantic, you must catch both
247	:class:`.DtypeError` and :class:`.ShapeError` (both of which are children
248	of :class:`.InterfaceError` )
249	"""
250	array = self.deserialize(array)	5✔
251
252	array = self.before_validation(array)	5✔
253
254	dtype = self.get_dtype(array)	5✔
255	dtype_valid = self.validate_dtype(dtype)	5✔
256	self.raise_for_dtype(dtype_valid, dtype)	5✔
257	array = self.after_validate_dtype(array)	5✔
258
259	shape = self.get_shape(array)	5✔
260	shape_valid = self.validate_shape(shape)	5✔
261	self.raise_for_shape(shape_valid, shape)	5✔
262
263	array = self.after_validation(array)	5✔
264
265	return array	5✔
266
267	def deserialize(self, array: Any) -> V \| Any:	5✔
268	"""
269	If given a JSON serialized version of the array,
270	deserialize it first.
271
272	If a roundtrip-serialized :class:`.JsonDict`,
273	pass to :meth:`.JsonDict.handle_input`.
274
275	If a roundtrip-serialized :class:`.MarkedJson`,
276	unpack mark, check for validity, warn if not,
277	and try to continue with validation
278	"""
279	if isinstance(marked_array := MarkedJson.try_cast(array), MarkedJson):	5✔
280	try:	5✔
281	marked_array.interface.is_valid(self.__class__, raise_on_error=True)	5✔
282	except MarkMismatchError as e:	5✔
283	warnings.warn(	5✔
284	str(e) + "\nAttempting to continue validation...", stacklevel=2
285	)
286	array = marked_array.value	5✔
287
288	return self.json_model.handle_input(array)	5✔
289
290	def before_validation(self, array: Any) -> NDArrayType:	5✔
291	"""
292	Optional step pre-validation that coerces the input into a type that can be
293	validated for shape and dtype
294
295	Default method is a no-op
296	"""
297	return array	×
298
299	def get_dtype(self, array: NDArrayType) -> DtypeType:	5✔
300	"""
301	Get the dtype from the input array.
302	"""
303	if hasattr(array.dtype, "type") and array.dtype.type is np.object_:	5✔
304	return self.get_object_dtype(array)	5✔
305	else:
306	return array.dtype	5✔
307
308	def get_object_dtype(self, array: NDArrayType) -> DtypeType:	5✔
309	"""
310	When an array contains an object, get the dtype of the object contained
311	by the array.
312
313	If this method returns `Any`, the dtype validation passes -
314	used for e.g. empty arrays for which the dtype of the array can't be determined
315	(since there are no objects).
316	"""
317	try:	5✔
318	return type(array.ravel()[0])	5✔
319	except IndexError:	5✔
320	return Any	5✔
321
322	def validate_dtype(self, dtype: DtypeType) -> bool:	5✔
323	"""
324	Validate the dtype of the given array, returning
325	``True`` if valid, ``False`` if not.
326	"""
327	return validate_dtype(dtype, self.dtype)	5✔
328
329	def raise_for_dtype(self, valid: bool, dtype: DtypeType) -> None:	5✔
330	"""
331	After validating, raise an exception if invalid
332	Raises:
333	:class:`~numpydantic.exceptions.DtypeError`
334	"""
335	if not valid:	5✔
336	raise DtypeError(f"Invalid dtype! expected {self.dtype}, got {dtype}")	5✔
337
338	def after_validate_dtype(self, array: NDArrayType) -> NDArrayType:	5✔
339	"""
340	Hook to modify array after validating dtype.
341	Default is a no-op.
342	"""
343	return array	5✔
344
345	def get_shape(self, array: NDArrayType) -> tuple[int, ...]:	5✔
346	"""
347	Get the shape from the array as a tuple of integers
348	"""
349	return array.shape	5✔
350
351	def validate_shape(self, shape: tuple[int, ...]) -> bool:	5✔
352	"""
353	Validate the shape of the given array against the shape
354	specifier, returning ``True`` if valid, ``False`` if not.
355
356
357	"""
358	if self.shape is Any:	5✔
359	return True	5✔
360
361	return validate_shape(shape, self.shape)	5✔
362
363	def raise_for_shape(self, valid: bool, shape: tuple[int, ...]) -> None:	5✔
364	"""
365	Raise a ShapeError if the shape is invalid.
366
367	Raises:
368	:class:`~numpydantic.exceptions.ShapeError`
369	"""
370	if not valid:	5✔
371	raise ShapeError(	5✔
372	f"Invalid shape! expected shape {self.shape.prepared_args}, "
373	f"got shape {shape}"
374	)
375
376	def after_validation(self, array: NDArrayType) -> T:	5✔
377	"""
378	Optional step post-validation that coerces the intermediate array type into the
379	return type
380
381	Default method is a no-op
382	"""
383	return array	5✔
384
385	@classmethod	5✔
386	@abstractmethod	5✔
387	def check(cls, array: Any) -> bool:	5✔
388	"""
389	Method to check whether a given input applies to this interface
390	"""
391
392	@classmethod	5✔
393	@abstractmethod	5✔
394	def enabled(cls) -> bool:	5✔
395	"""
396	Check whether this array interface can be used (eg. its dependent packages are
397	installed, etc.)
398	"""
399
400	@property	5✔
401	@abstractmethod	5✔
402	def name(self) -> str:	5✔
403	"""
404	Short name for this interface
405	"""
406
407	@property	5✔
408	@abstractmethod	5✔
409	def json_model(self) -> JsonDict:	5✔
410	"""
411	The :class:`.JsonDict` model used for roundtripping
412	JSON serialization
413	"""
414
415	@classmethod	5✔
416	@abstractmethod	5✔
417	def to_json(cls, array: type[T], info: SerializationInfo) -> list \| JsonDict:	5✔
418	"""
419	Convert an array of :attr:`.return_type` to a JSON-compatible format using
420	base python types
421	"""
422
423	@classmethod	5✔
424	def mark_json(cls, array: list \| dict) -> dict:	5✔
425	"""
426	When using ``model_dump_json`` with ``mark_interface: True`` in the ``context``,
427	add additional annotations that would allow the serialized array to be
428	roundtripped.
429
430	Default is just to add an :class:`.InterfaceMark`
431
432	Examples:
433
434	>>> from pprint import pprint
435	>>> pprint(Interface.mark_json([1.0, 2.0]))
436	{'interface': {'cls': 'Interface',
437	'module': 'numpydantic.interface.interface',
438	'version': '1.2.2'},
439	'value': [1.0, 2.0]}
440	"""
441	return {"interface": cls.mark_interface(), "value": array}	5✔
442
443	@classmethod	5✔
444	def interfaces(	5✔
445	cls, with_disabled: bool = False, sort: bool = True
446	) -> tuple[type["Interface"], ...]:
447	"""
448	Enabled interface subclasses
449
450	Args:
451	with_disabled (bool): If ``True`` , get every known interface.
452	If ``False`` (default), get only enabled interfaces.
453	sort (bool): If ``True`` (default), sort interfaces by priority.
454	If ``False`` , sorted by definition order. Used for recursion:
455	we only want to sort once at the top level.
456	"""
457	# get recursively
458	subclasses = []	5✔
459	for i in cls.__subclasses__():	5✔
460	if with_disabled:	5✔
461	subclasses.append(i)	5✔
462
463	if i.enabled():	5✔
464	subclasses.append(i)	5✔
465
466	subclasses.extend(i.interfaces(with_disabled=with_disabled, sort=False))	5✔
467
468	if sort:	5✔
469	subclasses = sorted(	5✔
470	subclasses,
471	key=attrgetter("priority"),
472	reverse=True,
473	)
474
475	return tuple(subclasses)	5✔
476
477	@classmethod	5✔
478	def return_types(cls) -> tuple[NDArrayType, ...]:	5✔
479	"""Return types for all enabled interfaces"""
480	return tuple([i.return_type for i in cls.interfaces()])	5✔
481
482	@classmethod	5✔
483	def input_types(cls) -> tuple[Any, ...]:	5✔
484	"""Input types for all enabled interfaces"""
485	in_types = []	5✔
486	for iface in cls.interfaces():	5✔
487	if isinstance(iface.input_types, (tuple, list)):	5✔
488	in_types.extend(iface.input_types)	5✔
489	else: # pragma: no cover
490	in_types.append(iface.input_types)
491
492	return tuple(in_types)	5✔
493
494	@classmethod	5✔
495	def match_mark(cls, array: Any) -> type["Interface"] \| None:	5✔
496	"""
497	Match a marked JSON dump of this array to the interface that it indicates.
498
499	First find an interface that matches by name, and then run its
500	``check`` method, because arrays can be dumped with a mark
501	but without ``round_trip == True`` (and thus can't necessarily
502	use the same interface that they were dumped with)
503
504	Returns:
505	Interface if match found, None otherwise
506	"""
507	mark = MarkedJson.try_cast(array)	5✔
508	if not isinstance(mark, MarkedJson):	5✔
509	return None	5✔
510
511	interface = mark.interface.match_by_name()	5✔
512	if interface is not None and interface.check(mark.value):	5✔
513	return interface	5✔
514	return None	×
515
516	@classmethod	5✔
517	def match(cls, array: Any, fast: bool = False) -> type["Interface"]:	5✔
518	"""
519	Find the interface that should be used for this array based on its input type
520
521	First runs the ``check`` method for all interfaces returned by
522	:meth:`.Interface.interfaces` except for :class:`.NumpyInterface` ,
523	and if no match is found then try the numpy interface. This is because
524	:meth:`.NumpyInterface.check` can be expensive, as we could potentially
525	try to
526
527	Args:
528	fast (bool): if ``False`` , check all interfaces and raise exceptions for
529	having multiple matching interfaces (default). If ``True`` ,
530	check each interface (as ordered by its ``priority`` , decreasing),
531	and return on the first match.
532	"""
533	# Shortcircuit match if this is a marked json dump
534	array = MarkedJson.try_cast(array)	5✔
535	if (match := cls.match_mark(array)) is not None:	5✔
536	return match	5✔
537	elif isinstance(array, MarkedJson):	5✔
538	array = array.value	×
539
540	# first try and find a non-numpy interface, since the numpy interface
541	# will try and load the array into memory in its check method
542	interfaces = cls.interfaces()	5✔
543	non_np_interfaces = [i for i in interfaces if i.name != "numpy"]	5✔
544	np_interface = [i for i in interfaces if i.name == "numpy"][0]	5✔
545
546	if fast:	5✔
547	matches = []	5✔
548	for i in non_np_interfaces:	5✔
549	if i.check(array):	5✔
550	return i	5✔
551	else:
552	matches = [i for i in non_np_interfaces if i.check(array)]	5✔
553
554	if len(matches) > 1:	5✔
555	msg = f"More than one interface matches input {array}:\n"	5✔
556	msg += "\n".join([f" - {i}" for i in matches])	5✔
557	raise TooManyMatchesError(msg)	5✔
558	elif len(matches) == 0:	5✔
559	# now try the numpy interface
560	if np_interface.check(array):	5✔
561	return np_interface	5✔
562	else:
563	raise NoMatchError(f"No matching interfaces found for input {array}")	5✔
564	else:
565	return matches[0]	5✔
566
567	@classmethod	5✔
568	def match_output(cls, array: Any) -> type["Interface"]:	5✔
569	"""
570	Find the interface that should be used based on the output type -
571	in the case that the output type differs from the input type, eg.
572	the HDF5 interface, match an instantiated array for purposes of
573	serialization to json, etc.
574	"""
575	matches = [i for i in cls.interfaces() if isinstance(array, i.return_type)]	5✔
576	if len(matches) > 1:	5✔
577	msg = f"More than one interface matches output {array}:\n"	5✔
578	msg += "\n".join([f" - {i}" for i in matches])	5✔
579	raise TooManyMatchesError(msg)	5✔
580	elif len(matches) == 0:	5✔
581	raise NoMatchError(f"No matching interfaces found for output {array}")	5✔
582	else:
583	return matches[0]	5✔
584
585	@classmethod	5✔
586	@lru_cache(maxsize=32)	5✔
587	def mark_interface(cls) -> InterfaceMark:	5✔
588	"""
589	Create an interface mark indicating this interface for validation after
590	JSON serialization with ``round_trip==True``
591	"""
592	interface_module = inspect.getmodule(cls)	5✔
593	interface_module = (	5✔
594	None if interface_module is None else interface_module.__name__
595	)
596	try:	5✔
597	v = (	5✔
598	None
599	if interface_module is None
600	else version(interface_module.split(".")[0])
601	)
602	except (
603	PackageNotFoundError
604	): # pragma: no cover - no tests for missing interface deps
605	v = None
606
607	return InterfaceMark(	5✔
608	module=interface_module, cls=cls.__name__, name=cls.name, version=v
609	)
610
611
612	class Proxy(ABC):	5✔
613	"""
614	A proxy class that exposes some non-array data source (like a video) as an array
615	"""
616
617	@classmethod	5✔
618	@abstractmethod	5✔
619	def proxy_for(cls) -> type[Interface]:	5✔
620	"""
621	Declare the interface that this is a proxy for,
622	allowing the proxy to be used with the NDArraySchema annotation
623	with any of the input types that the Interface supports.
624	"""
NEW 625	raise NotImplementedError()	×

p2p-ld / numpydantic / 26861324571

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