Skip to content
Closed
Show file tree
Hide file tree
Changes from 4 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
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file.
- Add an `encoding` kwarg on `MetricReader.open` and `MetricWriter.open` (#43)
- Transparent gzip/bz2/xz support via `xopen` (#44)
- Infer the delimiter from the file extension in `MetricReader.open`/`MetricWriter.open`; pass `delimiter=` to override (#61)
- Restrict boolean parsing on `Metric` to a narrow, configurable token set (the `BoolTokens` converter), defaulting to fgpyo's `{true, t, 1}`/`{false, f, 0}` and rejecting anything else (#75)

### Changed

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


def is_bool(annotation: TypeAnnotation | None) -> bool:
"""
Check if a type annotation is `bool` or an Optional `bool`.

Matches `bool`, `Optional[bool]`, and `bool | None`. Although `bool` is a subclass of `int`,
`int` is not matched. A union pairing `bool` with another non-None type (e.g. `bool | str`) is
not matched either, since it is not a pure boolean field.

Examples:
>>> is_bool(bool)
True
>>> is_bool(bool | None)
True
>>> is_bool(int)
False
>>> is_bool(bool | str)
False
>>> is_bool(list[bool])
False
"""
if annotation is None:
return False

if is_optional(annotation):
annotation = unpack_optional(annotation)

return annotation is bool


def is_counter(annotation: TypeAnnotation | None) -> bool:
"""
True if the type annotation is a Counter.
Expand Down
7 changes: 0 additions & 7 deletions fgmetric/collections/__init__.py

This file was deleted.

11 changes: 11 additions & 0 deletions fgmetric/converters/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
from fgmetric.converters._bool_tokens import BoolTokens
from fgmetric.converters._counter_pivot_table import CounterPivotTable
from fgmetric.converters._delimited_list import DelimitedList
from fgmetric.converters._null_sentinels import NullSentinels

__all__ = [
"BoolTokens",
"CounterPivotTable",
"DelimitedList",
"NullSentinels",
]
124 changes: 124 additions & 0 deletions fgmetric/converters/_bool_tokens.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
from typing import Any
from typing import ClassVar
from typing import final

from pydantic import BaseModel
from pydantic import ValidationInfo
from pydantic import field_validator

from fgmetric._typing_extensions import is_bool


# NB: Inheriting from BaseModel is necessary to declare field validators on the mixin, and for the
# class-level initialization in `__pydantic_init_subclass__` to work.
class BoolTokens(BaseModel):
"""
Restrict boolean parsing to a narrow, configurable set of tokens.

Pydantic accepts a broad set of strings as booleans (`true`/`t`/`yes`/`y`/`on`/`1` and their
false counterparts). With this mixin, fields annotated as `bool` or `bool | None` are instead
resolved from a narrow set of tokens *before* Pydantic's coercion runs: the incoming string
must match one of `true_tokens` or `false_tokens` (case-insensitively), or a `ValidationError`
is raised at the offending cell. `Metric` includes this behavior by default.

The default token set mirrors `fgpyo.util.types.parse_bool`: `{"true", "t", "1"}` for `True`
and `{"false", "f", "0"}` for `False`. This is stricter than Pydantic, which would also accept
`yes`/`y`/`on`/`no`/`n`/`off`. Rejecting those guards against a misaligned or wrong-type column
being silently coerced, and gives tools ported from fgpyo identical accept/reject behavior.

This mixin is read-side only; a `bool` still serializes to `"True"`/`"False"`.

Tokens are matched case-insensitively. `true_tokens` and `false_tokens` must be disjoint;
declaring a token in both is rejected at class definition.
Comment thread
msto marked this conversation as resolved.

Class Variables:
true_tokens: The strings resolved to `True`. Defaults to `frozenset({"true", "t", "1"})`.
false_tokens: The strings resolved to `False`. Defaults to `frozenset({"false", "f", "0"})`.

Note:
`bool` is a subclass of `int`, but only fields annotated exactly as `bool` or `bool | None`
are affected; `int` fields are untouched.

When combined with `NullSentinels`, the model-level null substitution runs first, so an
empty `bool | None` cell is converted to `None` before this validator sees it (provided
`""` is a configured null sentinel). Without such a sentinel, an empty cell on a `bool`
field is rejected, since `""` is in neither token set.
Comment thread
msto marked this conversation as resolved.

Examples:
`Metric` parses bool fields from the default tokens out of the box:

```python
class MyMetric(Metric):
flag: bool

MyMetric.model_validate({"flag": "1"}).flag # -> True
MyMetric.model_validate({"flag": "yes"}) # -> ValidationError
```

Customize the accepted tokens:

```python
class MyMetric(Metric):
true_tokens = frozenset({"yes", "y"})
false_tokens = frozenset({"no", "n"})
flag: bool
```
Comment thread
msto marked this conversation as resolved.
"""

# Public, user-specified token sets.
true_tokens: ClassVar[frozenset[str]] = frozenset({"true", "t", "1"})
false_tokens: ClassVar[frozenset[str]] = frozenset({"false", "f", "0"})

# Internal: lowercase copies of the user-specified token sets, used for the actual matching.
_true_tokens: ClassVar[frozenset[str]]
_false_tokens: ClassVar[frozenset[str]]

_bool_fieldnames: ClassVar[set[str]]

@classmethod
def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
"""
Record bool fields and the case-folded token sets used for matching.

1. The names of all fields annotated as `bool` or `bool | None` are stored in the private
`_bool_fieldnames` class variable.
2. `true_tokens` and `false_tokens` are case-folded into private sets for case-insensitive
matching, and validated to be disjoint.
"""
super().__pydantic_init_subclass__(**kwargs)

cls._bool_fieldnames = {
name for name, info in cls.model_fields.items() if is_bool(info.annotation)
}
cls._true_tokens = frozenset(token.casefold() for token in cls.true_tokens)
cls._false_tokens = frozenset(token.casefold() for token in cls.false_tokens)
cls._require_disjoint_tokens()

@classmethod
def _require_disjoint_tokens(cls) -> None:
"""Require that no token is configured as both true and false."""
overlap = cls._true_tokens & cls._false_tokens
if overlap:
raise ValueError(
"true_tokens and false_tokens must be disjoint,"
f" got overlapping tokens: {sorted(overlap)}"
)

@final
@field_validator("*", mode="before")
@classmethod
def _parse_bool_tokens(cls, value: Any, info: ValidationInfo) -> Any:
"""Resolve string values on bool fields to `True`/`False` from the configured tokens."""
if isinstance(value, str) and info.field_name in cls._bool_fieldnames:
token = value.casefold()
if token in cls._true_tokens:
return True
elif token in cls._false_tokens:
return False
else:
raise ValueError(
f"Invalid boolean token {value!r}; expected one of"
f" {sorted(cls.true_tokens | cls.false_tokens)}"
)

return value
116 changes: 116 additions & 0 deletions fgmetric/converters/_null_sentinels.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
from typing import Any
from typing import ClassVar
from typing import final

from pydantic import AliasChoices
from pydantic import BaseModel
from pydantic import model_validator
from pydantic.fields import FieldInfo

from fgmetric._typing_extensions import is_optional


def _validation_keys(info: FieldInfo) -> set[str]:
"""
Return the string keys an input field may be supplied under.

Covers the field's plain string `alias`/`validation_alias` and the string members of an
`AliasChoices`. `serialization_alias` is an output-only key, so it is excluded. `AliasPath`
aliases are path-shaped (for nested input) and do not apply to flat delimited rows, so they
are skipped.
"""
keys: set[str] = set()
if isinstance(info.alias, str):
keys.add(info.alias)
validation_alias = info.validation_alias
if isinstance(validation_alias, str):
keys.add(validation_alias)
elif isinstance(validation_alias, AliasChoices):
keys.update(choice for choice in validation_alias.choices if isinstance(choice, str))
return keys


# NB: Inheriting from BaseModel is necessary to declare model validators on the mixin, and for the
# class-level initialization in `__pydantic_init_subclass__` to work.
class NullSentinels(BaseModel):
"""
Treat configured input strings as null on Optional fields.

When this mixin is added to a model, the `null_sentinels` class variable declares the set of
input strings that represent null. Any field annotated as `T | None` whose incoming value is
one of the configured sentinels will be substituted with `None` before downstream field
validators run.

The substitution is scoped to Optional fields. Non-Optional fields are not touched, which
keeps this mixin composable with field-specific handling (e.g., `DelimitedList`'s treatment
of `list[T]` fields).

Class Variables:
null_sentinels: The set of input strings that should be treated as null on Optional
fields. Defaults to an empty set (no substitution).

Examples:
Treat empty strings as null:

```python
class MyMetric(Metric):
null_sentinels = frozenset({""})
name: str
count: int | None

MyMetric.model_validate({"name": "foo", "count": ""}).count # -> None
```

Treat multiple sentinels as null:

```python
class MyMetric(Metric):
null_sentinels = frozenset({"", "NA"})
count: int | None

MyMetric.model_validate({"count": "NA"}).count # -> None
```
"""

null_sentinels: ClassVar[frozenset[str]] = frozenset()
_optional_field_keys: ClassVar[set[str]]

@classmethod
def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
"""
Record the input keys of Optional-typed fields for null-sentinel substitution.

The `mode="before"` validator sees input keyed as it arrives from the file, which is the
field's alias when one is set. So for each Optional field we register its canonical name
and any alias it may appear under, ensuring aliased fields are recognized.
"""
super().__pydantic_init_subclass__(**kwargs)
keys: set[str] = set()
for name, info in cls.model_fields.items():
if not is_optional(info.annotation):
continue
keys.add(name)
keys |= _validation_keys(info)
cls._optional_field_keys = keys

@final
@model_validator(mode="before")
@classmethod
def _substitute_null_sentinels(cls, data: Any) -> Any:
"""Substitute incoming sentinel strings on Optional fields with `None`."""
if not cls.null_sentinels:
return data

if not isinstance(data, dict):
return data

data = dict(data)

for key in cls._optional_field_keys:
if key not in data:
continue
value = data[key]
if isinstance(value, str) and value in cls.null_sentinels:
data[key] = None

return data
Loading