24373313656

Committed 14 Apr 2026 12:00AM UTC coverage: 97.821% (-0.5%) from 98.345%

Build # 24373313656

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #67 from p2p-ld/ndarray-schema-proxies

allow proxies to be used with ndarray schema as input

Coverage Stats

15 of 17 new or added lines in 4 files covered. (88.24%)

23 existing lines in 11 files now uncovered.

1571 of 1606 relevant lines covered (97.82%)

9.78 hits per line

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

92.21

/src/numpydantic/serialization.py

"""
Serialization helpers for :func:`pydantic.BaseModel.model_dump`
and :func:`pydantic.BaseModel.model_dump_json` .
"""

from collections.abc import Callable, Iterable
from pathlib import Path
from typing import Any, TypeVar

from pydantic_core.core_schema import SerializationInfo

from numpydantic.interface import Interface, JsonDict

T = TypeVar("T")
U = TypeVar("U")


def jsonize_array(value: Any, info: SerializationInfo) -> list | dict:
    """Use an interface class to render an array as JSON"""
    # perf: keys to skip in generation - anything named "value" is array data.
    skip = ["value"]

    interface_cls = Interface.match_output(value)
    array = interface_cls.to_json(value, info)
    if isinstance(array, JsonDict):
        array = array.model_dump(exclude_none=True)

    if info.context:
        if info.context.get("mark_interface", False):
            array = interface_cls.mark_json(array)

        if isinstance(array, list):
            return array

        # ---- Perf Barrier ------------------------------------------------------
        # put context args intended to **wrap** the array above
        # put context args intended to **modify** the array below
        #
        # above, we assume that a list is **data** not to be modified.
        # below, we must mark whenever the data is in the line of fire
        # to avoid an expensive iteration.

        if info.context.get("absolute_paths", False):
            array = _absolutize_paths(array, skip)
        else:
            relative_to = info.context.get("relative_to", ".")
            array = _relativize_paths(array, relative_to, skip)
    else:
        if isinstance(array, list):
            return array

        # ---- Perf Barrier ------------------------------------------------------
        # same as above, ensure any keys that contain array values are skipped right now

        array = _relativize_paths(array, ".", skip)

    return array


def _relativize_paths(
    value: dict, relative_to: str = ".", skip: Iterable = tuple()
) -> dict:
    """
    Make paths relative to either the current directory or the provided
    ``relative_to`` directory, if provided in the context
    """
    relative_to = Path(relative_to).resolve()

    def _r_path(v: Any) -> Any:
        if not isinstance(v, (str, Path)):
            return v
        try:
            path = Path(v)
            resolved = path.resolve()
            # skip things that are pathlike but either don't exist
            # or that are at the filesystem root (eg like /data)
            if (
                not path.exists()
                or (resolved.is_dir() and str(resolved.parent) == resolved.anchor)
                or relative_to.anchor != resolved.anchor
            ):
                return v
            return str(relative_path(path, relative_to))
        except (TypeError, ValueError, OSError):
            return v

    return _walk_and_apply(value, _r_path, skip)


def _absolutize_paths(value: dict, skip: Iterable = tuple()) -> dict:
    def _a_path(v: Any) -> Any:
        if not isinstance(v, (str, Path)):
            return v
        try:
            path = Path(v)
            if not path.exists():
                return v
            return str(path.resolve())
        except (TypeError, ValueError, OSError):
            return v

    return _walk_and_apply(value, _a_path, skip)


def _walk_and_apply(value: T, f: Callable[[U, bool], U], skip: Iterable = tuple()) -> T:
    """
    Walk an object, applying a function
    """
    if isinstance(value, dict):
        for k, v in value.items():
            if k in skip:
                continue
            if isinstance(v, dict):
                _walk_and_apply(v, f, skip)
            elif isinstance(v, list):
                value[k] = [_walk_and_apply(sub_v, f, skip) for sub_v in v]
            else:
                value[k] = f(v)
    elif isinstance(value, list):
        value = [_walk_and_apply(v, f, skip) for v in value]
    else:
        value = f(value)
    return value


def relative_path(self: Path, other: Path, walk_up: bool = True) -> Path:
    """
    "Backport" of :meth:`pathlib.Path.relative_to` with ``walk_up=True``
    that's not available pre 3.12.

    Return the relative path to another path identified by the passed
    arguments.  If the operation is not possible (because this is not
    related to the other path), raise ValueError.

    The *walk_up* parameter controls whether `..` may be used to resolve
    the path.

    References:
        https://github.com/python/cpython/blob/8a2baedc4bcb606da937e4e066b4b3a18961cace/Lib/pathlib/_abc.py#L244-L270
    """
    if not isinstance(other, Path):  # pragma: no cover - ripped from cpython
        other = Path(other)
    self_parts = self.parts
    other_parts = other.parts
    anchor0, parts0 = self_parts[0], list(reversed(self_parts[1:]))
    anchor1, parts1 = other_parts[0], list(reversed(other_parts[1:]))
    if anchor0 != anchor1:
        raise ValueError(f"{self!r} and {other!r} have different anchors")
    while parts0 and parts1 and parts0[-1] == parts1[-1]:
        parts0.pop()
        parts1.pop()
    for part in parts1:  # pragma: no cover - not testing, ripped off from cpython
        if not part or part == ".":
            pass
        elif not walk_up:
            raise ValueError(f"{self!r} is not in the subpath of {other!r}")
        elif part == "..":
            raise ValueError(f"'..' segment in {other!r} cannot be walked")
        else:
            parts0.append("..")
    return Path(*reversed(parts0))

1	"""
2	Serialization helpers for :func:`pydantic.BaseModel.model_dump`
3	and :func:`pydantic.BaseModel.model_dump_json` .
4	"""
5
6	from collections.abc import Callable, Iterable	10✔
7	from pathlib import Path	10✔
8	from typing import Any, TypeVar	10✔
9
10	from pydantic_core.core_schema import SerializationInfo	10✔
11
12	from numpydantic.interface import Interface, JsonDict	10✔
13
14	T = TypeVar("T")	10✔
15	U = TypeVar("U")	10✔
16
17
18	def jsonize_array(value: Any, info: SerializationInfo) -> list \| dict:	10✔
19	"""Use an interface class to render an array as JSON"""
20	# perf: keys to skip in generation - anything named "value" is array data.
21	skip = ["value"]	10✔
22
23	interface_cls = Interface.match_output(value)	10✔
24	array = interface_cls.to_json(value, info)	10✔
25	if isinstance(array, JsonDict):	10✔
26	array = array.model_dump(exclude_none=True)	10✔
27
28	if info.context:	10✔
29	if info.context.get("mark_interface", False):	10✔
30	array = interface_cls.mark_json(array)	10✔
31
32	if isinstance(array, list):	10✔
33	return array	10✔
34
35	# ---- Perf Barrier ------------------------------------------------------
36	# put context args intended to wrap the array above
37	# put context args intended to modify the array below
38	#
39	# above, we assume that a list is data not to be modified.
40	# below, we must mark whenever the data is in the line of fire
41	# to avoid an expensive iteration.
42
43	if info.context.get("absolute_paths", False):	10✔
44	array = _absolutize_paths(array, skip)	10✔
45	else:
46	relative_to = info.context.get("relative_to", ".")	10✔
47	array = _relativize_paths(array, relative_to, skip)	10✔
48	else:
49	if isinstance(array, list):	10✔
50	return array	10✔
51
52	# ---- Perf Barrier ------------------------------------------------------
53	# same as above, ensure any keys that contain array values are skipped right now
54
55	array = _relativize_paths(array, ".", skip)	10✔
56
57	return array	10✔
58
59
60	def _relativize_paths(	10✔
61	value: dict, relative_to: str = ".", skip: Iterable = tuple()
62	) -> dict:
63	"""
64	Make paths relative to either the current directory or the provided
65	``relative_to`` directory, if provided in the context
66	"""
67	relative_to = Path(relative_to).resolve()	10✔
68
69	def _r_path(v: Any) -> Any:	10✔
70	if not isinstance(v, (str, Path)):	10✔
71	return v	10✔
72	try:	10✔
73	path = Path(v)	10✔
74	resolved = path.resolve()	10✔
75	# skip things that are pathlike but either don't exist
76	# or that are at the filesystem root (eg like /data)
77	if (	10✔
78	not path.exists()
79	or (resolved.is_dir() and str(resolved.parent) == resolved.anchor)
80	or relative_to.anchor != resolved.anchor
81	):
82	return v	10✔
83	return str(relative_path(path, relative_to))	10✔
UNCOV 84	except (TypeError, ValueError, OSError):	×
UNCOV 85	return v	×
86
87	return _walk_and_apply(value, _r_path, skip)	10✔
88
89
90	def _absolutize_paths(value: dict, skip: Iterable = tuple()) -> dict:	10✔
91	def _a_path(v: Any) -> Any:	10✔
92	if not isinstance(v, (str, Path)):	10✔
93	return v	10✔
94	try:	10✔
95	path = Path(v)	10✔
96	if not path.exists():	10✔
97	return v	10✔
98	return str(path.resolve())	10✔
99	except (TypeError, ValueError, OSError):	×
UNCOV 100	return v	×
101
102	return _walk_and_apply(value, _a_path, skip)	10✔
103
104
105	def _walk_and_apply(value: T, f: Callable[[U, bool], U], skip: Iterable = tuple()) -> T:	10✔
106	"""
107	Walk an object, applying a function
108	"""
109	if isinstance(value, dict):	10✔
110	for k, v in value.items():	10✔
111	if k in skip:	10✔
112	continue	10✔
113	if isinstance(v, dict):	10✔
114	_walk_and_apply(v, f, skip)	10✔
115	elif isinstance(v, list):	10✔
116	value[k] = [_walk_and_apply(sub_v, f, skip) for sub_v in v]	10✔
117	else:
118	value[k] = f(v)	10✔
119	elif isinstance(value, list):	10✔
UNCOV 120	value = [_walk_and_apply(v, f, skip) for v in value]	×
121	else:
122	value = f(value)	10✔
123	return value	10✔
124
125
126	def relative_path(self: Path, other: Path, walk_up: bool = True) -> Path:	10✔
127	"""
128	"Backport" of :meth:`pathlib.Path.relative_to` with ``walk_up=True``
129	that's not available pre 3.12.
130
131	Return the relative path to another path identified by the passed
132	arguments. If the operation is not possible (because this is not
133	related to the other path), raise ValueError.
134
135	The walk_up parameter controls whether `..` may be used to resolve
136	the path.
137
138	References:
139	https://github.com/python/cpython/blob/8a2baedc4bcb606da937e4e066b4b3a18961cace/Lib/pathlib/_abc.py#L244-L270
140	"""
141	if not isinstance(other, Path): # pragma: no cover - ripped from cpython
142	other = Path(other)
143	self_parts = self.parts	10✔
144	other_parts = other.parts	10✔
145	anchor0, parts0 = self_parts[0], list(reversed(self_parts[1:]))	10✔
146	anchor1, parts1 = other_parts[0], list(reversed(other_parts[1:]))	10✔
147	if anchor0 != anchor1:	10✔
UNCOV 148	raise ValueError(f"{self!r} and {other!r} have different anchors")	×
149	while parts0 and parts1 and parts0[-1] == parts1[-1]:	10✔
150	parts0.pop()	10✔
151	parts1.pop()	10✔
152	for part in parts1: # pragma: no cover - not testing, ripped off from cpython
153	if not part or part == ".":
154	pass
155	elif not walk_up:
156	raise ValueError(f"{self!r} is not in the subpath of {other!r}")
157	elif part == "..":
158	raise ValueError(f"'..' segment in {other!r} cannot be walked")
159	else:
160	parts0.append("..")
161	return Path(*reversed(parts0))	10✔

p2p-ld / numpydantic / 24373313656

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