Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ All notable changes to this project will be documented in this file.
- `MetricWriter` is IO-first; open a path with `MetricWriter.open()` rather than passing it to the constructor (#42)
- `Metric.read` is now a thin wrapper over `MetricReader.open` and reads eagerly, returning a `list` instead of a lazy generator; it accepts `str` paths in addition to `Path` and gains transparent decompression and the `encoding` kwarg. Use `MetricReader.open` to stream metrics without holding them all in memory (#62)
- `MetricReader.open` and `MetricWriter.open` eagerly validate that the path is readable/writable on context entry (#66)
- Rename the `DelimitedList` converter to `DelimitedCollection` (#76)
- Generalize the `DelimitedList` converter to `DelimitedCollection`, which now also handles `set`, `frozenset`, and `tuple` (including heterogeneous tuples such as `tuple[int, str]`) in addition to `list`. Set and frozenset output is sorted by serialized form for stable roundtrips (#76)

## [0.3.0] - 2026-05-12

Expand Down
27 changes: 27 additions & 0 deletions fgmetric/_typing_extensions.py
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,33 @@ def is_list(annotation: TypeAnnotation | None) -> bool:
return has_origin(annotation, list)


def is_collection(annotation: TypeAnnotation | None) -> bool:
"""
Check if a type annotation is a delimited-collection type.

Matches parameterized `list`, `set`, `frozenset`, and `tuple` (including the optional and
higher-arity tuple forms). Mappings (`dict`, `Counter`) and bare, unparameterized collections
are not collections for this purpose.

Examples:
>>> is_collection(list[int])
True
>>> is_collection(set[str])
True
>>> is_collection(frozenset[int])
True
>>> is_collection(tuple[int, str])
True
>>> is_collection(tuple[int, ...] | None)
True
>>> is_collection(dict[str, int])
False
>>> is_collection(list) # bare list, no type parameter
False
"""
return any(has_origin(annotation, origin) for origin in (list, set, frozenset, tuple))


def is_counter(annotation: TypeAnnotation | None) -> bool:
"""
True if the type annotation is a Counter.
Expand Down
205 changes: 143 additions & 62 deletions fgmetric/converters/_delimited_collection.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
from typing import Any
from typing import ClassVar
from typing import TypeAlias
from typing import final
from typing import get_args
from typing import get_origin

from pydantic import BaseModel
from pydantic import FieldSerializationInfo
Expand All @@ -10,35 +11,43 @@
from pydantic import field_serializer
from pydantic import field_validator

from fgmetric._typing_extensions import has_optional_elements
from fgmetric._typing_extensions import is_list

Fieldname: TypeAlias = str
"""A pydantic model field's name."""
from fgmetric._typing_extensions import is_collection
from fgmetric._typing_extensions import is_optional
from fgmetric._typing_extensions import unpack_optional


# NB: Inheriting from BaseModel is necessary to declare field/model validators on the mixin, and
# for the class-level validations defined in `__pydantic_init_subclass__` to work.
class DelimitedCollection(BaseModel):
"""
Serialize and deserialize delimited lists of (de)serializable types.
Serialize and deserialize delimited collections of (de)serializable types.

When this mixin is added to `Metric`, fields annotated as `list[T]` will be read and written as
comma-delimited strings. During validation, a comma-delimited string will be split into a list
and its elements validated as instances of `T`. During serialization, the list elements will be
serialized to string and then joined into a comma-delimited string.
When this mixin is added to `Metric`, fields annotated as `list[T]`, `set[T]`, `frozenset[T]`,
or `tuple[...]` will be read and written as delimited strings. During validation, a delimited
string is split into its elements, which are validated as instances of the declared element
type(s); Pydantic performs the container coercion (list, set, frozenset, or tuple). During
serialization, the elements are serialized to string and joined back into a delimited string.

The list type `T` may be any serializable type. The field may be annotated as `list[T]` or
`list[T] | None` - as with any primitive type, `None` will be validated from and serialized to
the empty string.
The element type(s) may be any serializable type. A field may be annotated as `list[T]`,
`set[T]`, `frozenset[T]`, the variadic `tuple[T, ...]`, or the fixed-arity, heterogeneous
`tuple[T1, T2, ...]` (whose arity and per-position types Pydantic validates). Any of these may
also be made optional (e.g. `list[T] | None`); as with any primitive type, `None` is validated
from and serialized to the empty string.

The delimiter may be configured by specifying the `collection_delimiter` class variable when
declaring a model.

Note:
Roundtrips are lossy if list elements contain the delimiter character. For example, with the
`list` and `tuple` are ordered, so their element order is preserved. `set` and `frozenset`
are unordered (and string hashing is per-process salted), so their elements are serialized
sorted by their serialized form, which keeps the output stable across runs and roundtrips.

Note:
Roundtrips are lossy if elements contain the delimiter character. For example, with the
default comma delimiter, `["a,b", "c"]` serializes to `"a,b,c"` and deserializes back to
`["a", "b", "c"]`. Avoid using delimiters that may appear in element values.
`["a", "b", "c"]`. Avoid using delimiters that may appear in element values. Collections
are also flat: nested collections (e.g. `list[list[int]]`) are not supported, because a
flat delimited string cannot be unambiguously re-nested.

Examples:
Basic usage — comma delimiter (default):
Expand All @@ -51,6 +60,25 @@ class MyMetric(Metric):
MyMetric(tags=[1, 2, 3]).model_dump() # -> {"tags": "1,2,3"}
```

Sets and frozensets — output is sorted by serialized form:

```python
class MyMetric(Metric):
tags: set[int] # "3,1,2" becomes {1, 2, 3}

MyMetric.model_validate({"tags": "3,1,2"}).tags # -> {1, 2, 3}
MyMetric(tags={3, 1, 2}).model_dump() # -> {"tags": "1,2,3"}
```

Heterogeneous tuples — arity and per-position types are validated:

```python
class MyMetric(Metric):
point: tuple[int, str] # "1,foo" becomes (1, "foo")

MyMetric.model_validate({"point": "1,foo"}).point # -> (1, "foo")
```

Custom delimiter:

```python
Expand All @@ -62,7 +90,7 @@ class MyMetric(Metric):
MyMetric(tags=[1, 2, 3]).model_dump() # -> {"tags": "1;2;3"}
```

Optional list field — the whole field may be absent:
Optional collection field — the whole field may be absent:

```python
class MyMetric(Metric):
Expand All @@ -72,7 +100,7 @@ class MyMetric(Metric):
MyMetric(tags=None).model_dump() # -> {"tags": None}
```

List with optional elements — individual elements may be absent:
Collection with optional elements — individual elements may be absent:

```python
class MyMetric(Metric):
Expand All @@ -84,29 +112,23 @@ class MyMetric(Metric):
"""

collection_delimiter: ClassVar[str] = ","
_list_fieldnames: ClassVar[set[str]]
_optional_element_fieldnames: ClassVar[set[str]]
_collection_fieldnames: ClassVar[set[str]]
_uniform_optional_element_fieldnames: ClassVar[set[str]]
_tuple_optional_positions: ClassVar[dict[str, tuple[bool, ...]]]

@classmethod
def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
"""
Validations of the user-defined model.
Validate the collection delimiter and classify the model's collection fields.

1. The collection delimiter must be a single character.
2. The names of all fields annotated as `list[T]` or `list[T] | None` are stored in the
private `_list_fieldnames` class variable.
2. Each `list`/`set`/`frozenset`/`tuple` field is recorded, along with the
per-element/per-position optionality needed to map empty cells to `None`.
"""
super().__pydantic_init_subclass__(**kwargs)

cls._require_single_character_collection_delimiter()
cls._list_fieldnames = {
name for name, info in cls.model_fields.items() if is_list(info.annotation)
}
cls._optional_element_fieldnames = {
name
for name, info in cls.model_fields.items()
if has_optional_elements(info.annotation)
}
cls._classify_collection_fields()

@classmethod
def _require_single_character_collection_delimiter(cls) -> None:
Expand All @@ -117,51 +139,110 @@ def _require_single_character_collection_delimiter(cls) -> None:
f" got: {cls.collection_delimiter!r}"
)

@classmethod
def _classify_collection_fields(cls) -> None:
"""Scan the model's fields once, recording each collection field's behavior."""
cls._collection_fieldnames = set()
cls._uniform_optional_element_fieldnames = set()
cls._tuple_optional_positions = {}

for name, info in cls.model_fields.items():
annotation = info.annotation
if not is_collection(annotation):
continue

cls._collection_fieldnames.add(name)

inner = _strip_optional(annotation)
args = get_args(inner)

# A fixed-arity tuple (e.g. `tuple[int, str]`) has per-position element types; every
# other collection — including the variadic `tuple[T, ...]` — has a single, uniform
# element type.
is_fixed_tuple = (
get_origin(inner) is tuple and args != () and not _is_variadic_tuple_args(args)
)
if is_fixed_tuple:
cls._tuple_optional_positions[name] = tuple(is_optional(arg) for arg in args)
elif args and is_optional(args[0]):
cls._uniform_optional_element_fieldnames.add(name)

@final
@field_validator("*", mode="before")
@classmethod
def _split_lists(cls, value: Any, info: ValidationInfo) -> Any:
"""Split any fields annotated as `list[T]` on a comma delimiter."""
if isinstance(value, str) and cls._is_list_field(info.field_name):
if value:
value = value.split(cls.collection_delimiter)
def _split_collections(cls, value: Any, info: ValidationInfo) -> Any:
"""Split any collection field into the intermediate list Pydantic then coerces."""
if isinstance(value, str) and cls._is_collection_field(info.field_name):
return cls._split_collection(value, info.field_name)
return value

@final
@classmethod
def _split_collection(cls, value: str, name: str | None) -> list[Any]:
"""Split a collection cell into a flat list of (string-or-`None`) elements."""
if not value:
return []

# Convert empty strings to None for list[T | None] fields
if info.field_name in cls._optional_element_fieldnames:
value = [None if el == "" else el for el in value]
elements: list[Any] = value.split(cls.collection_delimiter)

else:
value = []
# Map empty elements to `None` where the corresponding element type is optional.
if name in cls._uniform_optional_element_fieldnames:
elements = [None if element == "" else element for element in elements]
elif name in cls._tuple_optional_positions:
mask = cls._tuple_optional_positions[name]
elements = [
None if (index < len(mask) and mask[index] and element == "") else element
for index, element in enumerate(elements)
]

return value
return elements

@final
@field_serializer("*", mode="wrap")
def _join_lists(
def _join_collections(
self,
value: Any,
nxt: SerializerFunctionWrapHandler, # noqa: ARG002
nxt: SerializerFunctionWrapHandler,
info: FieldSerializationInfo,
) -> Any:
"""Join any fields annotated as `list[T]` with a delimiter."""
if isinstance(value, list) and self._is_list_field(info.field_name):
# Let the default serializer handle each item first. This should return a list of
# serialized values, applying default serialization to each list element.
serialized_value = nxt(value)

if isinstance(serialized_value, list):
# If the handler returned a list, join it. (This is the expected branch.)
# Also serialize `None` back to empty string.
elements = ["" if item is None else str(item) for item in serialized_value]
return self.collection_delimiter.join(elements)
else:
# If the handler already serialized to something else (unlikely), return as-is.
return serialized_value

"""Join any collection field back into a single delimited string."""
if self._is_collection_field(info.field_name) and isinstance(
value, (list, set, frozenset, tuple)
):
return self._join_collection(value, nxt)
return nxt(value)

@final
def _join_collection(self, value: Any, nxt: SerializerFunctionWrapHandler) -> Any:
"""Serialize each element, then join into a delimited string."""
# Let the default serializer handle each element first, applying any per-element custom
# serialization. This returns the elements as an iterable (a list in JSON mode).
serialized = nxt(value)
if not isinstance(serialized, (list, set, frozenset, tuple)):
# If the handler already produced something else (unlikely), return it as-is.
return serialized

elements = ["" if item is None else str(item) for item in serialized]

# `list`/`tuple` are ordered, so preserve order. `set`/`frozenset` are unordered (and
# string hashing is per-process salted), so sort by serialized form for a stable roundtrip.
if isinstance(value, (set, frozenset)):
elements.sort()

return self.collection_delimiter.join(elements)

@final
@classmethod
def _is_list_field(cls, field_name: str | None) -> bool:
"""True if the field is annotated as `list[T]` on the class model."""
return field_name is not None and field_name in cls._list_fieldnames
def _is_collection_field(cls, field_name: str | None) -> bool:
"""True if the field is annotated as a delimited collection on the class model."""
return field_name is not None and field_name in cls._collection_fieldnames


def _strip_optional(annotation: Any) -> Any:
"""Return the inner type of an optional annotation, or the annotation unchanged."""
return unpack_optional(annotation) if is_optional(annotation) else annotation


def _is_variadic_tuple_args(args: tuple[Any, ...]) -> bool:
"""True if a tuple's type arguments describe a variadic `tuple[T, ...]`."""
return len(args) == 2 and args[1] is Ellipsis
8 changes: 4 additions & 4 deletions fgmetric/metric.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,12 +35,12 @@ class Metric(
1. **Null sentinels.** Empty strings in Optional fields are converted to `None` before field
validation. The sentinel values converted to `None` can be overridden by the
`null_sentinels` class variable.
2. **Delimited lists.** Any field typed as `list[T]` will be parsed from and serialized to a
delimited string. The list delimiter may be controlled by the `collection_delimiter` class
variable.
2. **Delimited collections.** Any field typed as `list[T]`, `set[T]`, `frozenset[T]`, or
`tuple[...]` will be parsed from and serialized to a delimited string. The delimiter may be
controlled by the `collection_delimiter` class variable.

Class Variables:
collection_delimiter: A single-character delimiter used to split and join `list` fields
collection_delimiter: A single-character delimiter used to split and join collection fields
during serialization/deserialization.
null_sentinels: The set of input strings that should be treated as `None` on Optional
fields during validation. Defaults to `frozenset({""})`.
Expand Down
Loading