call_report.core.FieldSchema#

class call_report.core.FieldSchema(*, fields: Iterable[FieldAttributes])[source]#

Bases: Mapping[str, FieldAttributes]

An ordered, immutable mapping of field name to FieldAttributes.

Preserves the order fields were supplied in. Since instances are immutable, subset and add_fields return new FieldSchema instances rather than modifying this one.

Parameters:

fieldsIterable[FieldAttributes]: The fields making up this schema, in order.

Raises:

SchemaError: If two or more fields in fields share the same name.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> uninum = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> schema = FieldSchema(fields=[uninum])
>>> schema["UNINUM"] is uninum
True

property names: tuple[str, ...][source]#

Return the field names in schema order.

A convenience for inspecting field order without iterating.

Returns:

tuple[str, …]: The field names, in the order fields were defined.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> field = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> FieldSchema(fields=[field]).names
('UNINUM',)

property schema: Schema[source]#

Return this schema’s fields as a narwhals Schema.

Built once at construction from each field’s name and its latest version’s dtype. subset, add_fields, as_of, and from_dataframe construct their result via FieldSchema(fields=…), which runs this same logic, so they all get an up to date narwhals Schema for free. The returned object is read-only: it rejects in-place mutation such as item assignment, so it cannot be used to change this schema’s fields.

Returns:

narwhals.Schema: An ordered mapping of field name to narwhals dtype.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> field = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> FieldSchema(fields=[field]).schema
Schema({'UNINUM': Int64})

subset(*, names: Iterable[str]) → FieldSchema[source]#

Return a new FieldSchema containing only the named fields.

The result preserves this schema’s existing field order. It does not reorder fields to match the order of names.

Parameters:

namesIterable[str]: The field names to keep.

Returns:

FieldSchema: A new schema containing only the requested fields.

Raises:

SchemaError: If any name in names is not a field in this schema.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> uninum = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> rssd = FieldAttributes(
...     name="RSSD",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> schema = FieldSchema(fields=[uninum, rssd])
>>> schema.subset(names=["UNINUM"]).names
('UNINUM',)

add_fields(*, fields: FieldAttributes | Iterable[FieldAttributes], index: int | None = None) → FieldSchema[source]#

Return a new FieldSchema with one or more fields inserted.

This schema is left unmodified. The caller receives a new instance.

Parameters:

fieldsFieldAttributes or Iterable[FieldAttributes]: The field(s) to add.
indexint, optional: The position to insert at, following list.insert position semantics. If omitted, the field(s) are appended at the end.

Returns:

FieldSchema: A new schema with fields inserted at index.

Raises:

SchemaError: If fields is empty, index is out of range, or the result would contain a duplicate field name.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> uninum = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> rssd = FieldAttributes(
...     name="RSSD",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> FieldSchema(fields=[uninum]).add_fields(fields=rssd, index=0).names
('RSSD', 'UNINUM')

as_of(*, period: str | date | ReportingPeriod) → FieldSchema[source]#

Return a new FieldSchema of only the fields present as of period.

Each surviving field is narrowed to a single version covering the quarter period, using whichever of that field’s versions was actually active at that date rather than its most recent one. That is what makes the snapshot historically accurate: a field whose definition was later revised still shows the definition that applied at period, not today’s.

Parameters:

periodstr, datetime.date, or ReportingPeriod: The quarter-end to take the snapshot at.

Returns:

FieldSchema: A new schema containing only fields present at period, each narrowed to a single version.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> field = FieldAttributes(
...     name="INV_CODE",
...     versions=(
...         FieldVersion(
...             dtype=nw.Int64(),
...             definition="old text",
...             periods=PeriodRange(start="2000-03-31", end="2014-12-31"),
...         ),
...         FieldVersion(
...             dtype=nw.Int64(),
...             definition="new text",
...             periods=PeriodRange(start="2015-03-31", end="2026-03-31"),
...         ),
...     ),
... )
>>> schema = FieldSchema(fields=[field])
>>> schema.as_of(period="2010-03-31")["INV_CODE"].versions[0].definition
'old text'
>>> schema.as_of(period="2020-03-31")["INV_CODE"].versions[0].definition
'new text'

is_equal(*, other: FieldSchema, check_order: bool = False) → bool[source]#

Return whether other defines the same fields and metadata.

Content comparison is always order-insensitive (inherited from Mapping.__eq__): two schemas with the same fields in a different order compare equal by default. Pass check_order=True to additionally require the fields appear in the same sequence.

Parameters:

otherFieldSchema: The schema to compare against.
check_orderbool, default False: If True, also require identical field order.

Returns:

bool: True if other has the same fields (and, if check_order is True, the same field order).

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> uninum = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> rssd = FieldAttributes(
...     name="RSSD",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> forward = FieldSchema(fields=[uninum, rssd])
>>> reordered = FieldSchema(fields=[rssd, uninum])
>>> forward.is_equal(other=reordered)
True
>>> forward.is_equal(other=reordered, check_order=True)
False

compare(*, other: FieldSchema, check_order: bool = False) → FieldSchemaDiff[source]#

Compare this schema against other, field by field.

Unlike is_equal, this returns the actual differences rather than a single bool, which is useful for reviewing what changed between two point-in-time snapshots or auditing a metadata regeneration run.

Parameters:

otherFieldSchema: The schema to compare against.
check_orderbool, default False: If True, also detect whether the fields common to both schemas appear in a different relative order.

Returns:

FieldSchemaDiff: The fields added, removed, and changed between the two schemas.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> def _field(name: str, definition: str) -> FieldAttributes:
...     return FieldAttributes(
...         name=name,
...         versions=(
...             FieldVersion(
...                 dtype=nw.Int64(), definition=definition, periods=span
...             ),
...         ),
...     )
>>> before = FieldSchema(fields=[_field("UNINUM", "old"), _field("RSSD", "")])
>>> after = FieldSchema(fields=[_field("UNINUM", "new"), _field("ASSOC", "")])
>>> diff = before.compare(other=after)
>>> diff.added
('ASSOC',)
>>> diff.removed
('RSSD',)
>>> [change.name for change in diff.changed]
['UNINUM']
>>> diff.is_empty
False

to_dataframe(*, backend: Literal['pandas', 'polars', 'pyarrow'] | None = None, dataframe_type: None = None) → NativeDataFrame[source]#

to_dataframe(*, backend: Literal['pandas', 'polars', 'pyarrow'] | None = None, dataframe_type: Literal['pandas']) → pandas.DataFrame

to_dataframe(*, backend: Literal['pandas', 'polars', 'pyarrow'] | None = None, dataframe_type: Literal['pyarrow_table']) → pyarrow.Table

to_dataframe(*, backend: Literal['pandas', 'polars', 'pyarrow'] | None = None, dataframe_type: Literal['polars_dataframe']) → polars.DataFrame

to_dataframe(*, backend: Literal['pandas', 'polars', 'pyarrow'] | None = None, dataframe_type: Literal['polars_lazyframe']) → polars.LazyFrame

Return this schema as a native dataframe, one row per field version.

A field with more than one version (redefined in place without a presence gap, dropped and later reintroduced, or both) contributes one row per version, each with that version’s own dtype and definition.

Parameters:

backend{“pandas”, “polars”, “pyarrow”}, optional: The dataframe library used to build the frame. If omitted, uses whatever backend is currently configured via call_report.config.get_config. Most users can leave this at its default.
dataframe_type{“pandas”, “pyarrow_table”, “polars_lazyframe”, “polars_dataframe”}, optional: The dataframe type to convert the result to as a final step, regardless of backend. Leave this None (the default) to get back whatever backend produced. Set it when the code that consumes this result needs a specific type, for example a pandas DataFrame while the package is configured to use polars. The conversion is zero-copy when the requested type already matches.

Returns:

NativeDataFrame: A native dataframe with columns field_name, dtype, definition, period_start, and period_end (the last two as ISO YYYY-MM-DD strings). dtype holds each version’s narwhals dtype as its repr (e.g. "Int64", "Datetime(time_unit='us', time_zone='UTC')"). Storing it as a plain string lets it round-trip through every backend, including pyarrow, which cannot hold arbitrary Python objects as column values.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> field = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> frame = FieldSchema(fields=[field]).to_dataframe()
>>> list(frame.columns)
['field_name', 'dtype', 'definition', 'period_start', 'period_end']

classmethod from_dataframe(*, data: Any) → FieldSchema[source]#

Reconstruct a FieldSchema from a dataframe built by to_dataframe.

Rows are grouped by field_name, so a field is reconstructed with all of its versions even if data has one row per version.

Parameters:

dataAny: A native dataframe with the columns to_dataframe produces.

Returns:

FieldSchema: The reconstructed schema.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> field = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> frame = FieldSchema(fields=[field]).to_dataframe()
>>> FieldSchema.from_dataframe(data=frame).names
('UNINUM',)

to_json(*, indent: int | None = 2) → str[source]#

Return this schema as a JSON string.

The format this package ships canonical schedule metadata in (see call_report.fca.get_fca_file_metadata). It is used in preference to a flat dataframe because a field’s versions nest naturally under its name.

Parameters:

indentint, optional: Passed through to json.dumps. The default of 2 produces human-readable, diffable output, matching the shipped metadata files. Pass None for the most compact representation.

Returns:

str: A JSON object mapping each field name to its versions list, field order preserved.

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> field = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> schema = FieldSchema(fields=[field])
>>> FieldSchema.from_json(text=schema.to_json()) == schema
True

classmethod from_json(*, text: str) → FieldSchema[source]#

Reconstruct a FieldSchema from JSON built by to_json.

The inverse of to_json. Round-tripping a schema through both reconstructs an equal FieldSchema.

Parameters:

textstr: A JSON string in the shape to_json produces.

Returns:

FieldSchema: The reconstructed schema.

Raises:

SchemaError: If text is not valid JSON, is valid JSON that isn’t a JSON object, or doesn’t otherwise match the shape to_json produces (a missing key, a bad dtype repr, or an invalid period value).

Examples

>>> import narwhals as nw
>>> from call_report.core import (
...     FieldAttributes,
...     FieldSchema,
...     FieldVersion,
...     PeriodRange,
... )
>>> span = PeriodRange(start="2000-03-31", end="2026-03-31")
>>> field = FieldAttributes(
...     name="UNINUM",
...     versions=(FieldVersion(dtype=nw.Int64(), definition="", periods=span),),
... )
>>> schema = FieldSchema(fields=[field])
>>> FieldSchema.from_json(text=schema.to_json()).names
('UNINUM',)