Skip to content

ftmq.statements

Cast statement values into the canonical format of their property type.

cast_types normalizes number and date values: the parsed value goes into value, the raw string into original_value (unless already set), and the statement id is regenerated. Statement stores apply this on write, as the SQL backends CAST the value column. Existing dumps are migrated via the CLI:

cat statements.csv | ftmq statements cast-types > statements.typed.csv

cast_date(value)

Get the canonical ISO (prefix) form of a date value, or None if it doesn't parse. Partial dates ("2021", "2021-06") are kept.

Source code in ftmq/statements.py 
35
36
37
38
def cast_date(value: str) -> str | None:
    """Get the canonical ISO (prefix) form of a date value, or `None` if it doesn't
    parse. Partial dates (`"2021"`, `"2021-06"`) are kept."""
    return registry.date.clean_text(value)

cast_number(value)

Get the canonical numeric form of a value, or None if it doesn't parse.

Kept as a string (no float precision loss); a unit suffix is dropped ("5 kg" -> "5").

Source code in ftmq/statements.py 
25
26
27
28
29
30
31
32
def cast_number(value: str) -> str | None:
    """Get the canonical numeric form of a value, or `None` if it doesn't parse.

    Kept as a string (no float precision loss); a unit suffix is dropped
    (`"5 kg"` -> `"5"`).
    """
    number, _unit = registry.number.parse(value)
    return number

cast_statement(stmt, casters=None)

Cast a statement's value into the canonical format of its property type.

Parameters:

Name Type Description Default
stmt Statement

The statement to cast

 required
casters dict[str, Caster] | None

Caster per property type (default: number, date)

 None

Returns:

Type Description
Statement | None

The statement, cloned if its value changed, or None if the value doesn't parse
Source code in ftmq/statements.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def cast_statement(
    stmt: Statement, casters: dict[str, Caster] | None = None
) -> Statement | None:
    """Cast a statement's value into the canonical format of its property type.

    Args:
        stmt: The statement to cast
        casters: Caster per property type (default: `number`, `date`)

    Returns:
        The statement, cloned if its value changed, or `None` if the value
            doesn't parse
    """
    caster = (CASTERS if casters is None else casters).get(stmt.prop_type)
    if caster is None:
        return stmt
    value = caster(stmt.value)
    if value is None:
        log.warning(
            f"Invalid `{stmt.prop_type}` value: `{stmt.value}`",
            entity_id=stmt.entity_id,
            prop=stmt.prop,
        )
        return None
    if value == stmt.value:
        return stmt
    # `clone` regenerates the id, which is a content hash over the value
    return stmt.clone(value=value, original_value=stmt.original_value or stmt.value)

cast_types(statements, types=DEFAULT_TYPES, drop_invalid=False)

Normalize the values of the given property types in a statement stream.

Parameters:

Name Type Description Default
statements Iterable[Statement]

Stream of statements

 required
types Iterable[str]

Property types to cast (default: number, date)

 DEFAULT_TYPES
drop_invalid bool

Drop statements whose value doesn't parse instead of passing them through unchanged

 False

Returns:

Type Description
Iterator[Statement]

A generator of Statement instances
Source code in ftmq/statements.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def cast_types(
    statements: Iterable[Statement],
    types: Iterable[str] = DEFAULT_TYPES,
    drop_invalid: bool = False,
) -> Iterator[Statement]:
    """Normalize the values of the given property types in a statement stream.

    Args:
        statements: Stream of statements
        types: Property types to cast (default: `number`, `date`)
        drop_invalid: Drop statements whose value doesn't parse instead of
            passing them through unchanged

    Returns:
        A generator of `Statement` instances
    """
    # validate eagerly, before a consumer has written any output
    return _cast_types(statements, get_casters(types), drop_invalid)

get_casters(types)

Get the caster per property type, keyed as Statement.prop_type.

Source code in ftmq/statements.py 
48
49
50
51
52
53
54
55
56
57
58
def get_casters(types: Iterable[str]) -> dict[str, Caster]:
    """Get the caster per property type, keyed as `Statement.prop_type`."""
    casters: dict[str, Caster] = {}
    for type_ in types:
        caster = CASTERS.get(type_)
        if caster is None:
            raise ValueError(
                f"Invalid cast type: `{type_}` - one of ({', '.join(DEFAULT_TYPES)})"
            )
        casters[type_] = caster
    return casters