Class tigrbl.column.io_spec.IOSpec

tigrbl.column.io_spec.IOSpec dataclass

IOSpec(
    in_verbs=(),
    out_verbs=(),
    mutable_verbs=(),
    alias_in=None,
    alias_out=None,
    header_in=None,
    header_out=None,
    header_required_in=False,
    sensitive=False,
    redact_last=None,
    filter_ops=(),
    sortable=False,
    allow_in=None,
    allow_out=None,
    _paired=None,
    _assemble=None,
    _readtime_aliases=(),
)

Control how a column participates in API input and output.

in_verbs and out_verbs enumerate which operations may accept or emit the field. alias_in/alias_out allow different payload keys, sensitive and redact_last mark values for masking, and filter_ops/sortable flag whether the field can be used in query filters or ordering. Advanced helpers like :meth:assemble, :meth:paired, and :meth:alias_readtime derive values or expose extra read-time aliases.

in_verbs class-attribute instance-attribute

in_verbs = ()

out_verbs class-attribute instance-attribute

out_verbs = ()

mutable_verbs class-attribute instance-attribute

mutable_verbs = ()

alias_in class-attribute instance-attribute

alias_in = None

alias_out class-attribute instance-attribute

alias_out = None

header_in class-attribute instance-attribute

header_in = None

header_out class-attribute instance-attribute

header_out = None

header_required_in class-attribute instance-attribute

header_required_in = False

sensitive class-attribute instance-attribute

sensitive = False

redact_last class-attribute instance-attribute

redact_last = None

filter_ops class-attribute instance-attribute

filter_ops = ()

sortable class-attribute instance-attribute

sortable = False

allow_in class-attribute instance-attribute

allow_in = None

allow_out class-attribute instance-attribute

allow_out = None

assemble

assemble(sources, fn)

Return a new spec that derives a value from sources using fn.

Source code in tigrbl/column/io_spec.py

def assemble(self, sources, fn):
    """Return a new spec that derives a value from ``sources`` using ``fn``."""
    cfg = _AssembleCfg(sources=tuple(sources), fn=fn)
    return replace(self, _assemble=cfg)

paired

paired(
    make,
    *,
    alias,
    verbs=("create",),
    emit="pre_flush",
    alias_field=FieldSpec(py_type=str),
    mask_last=None,
)

Return a new spec with a paired field configuration.

Source code in tigrbl/column/io_spec.py

def paired(
    self,
    make,
    *,
    alias,
    verbs=("create",),
    emit: EmitPoint = "pre_flush",
    alias_field: F = F(py_type=str),
    mask_last: int | None = None,
):
    """Return a new spec with a paired field configuration."""

    def gen(ctx):
        pair = make(ctx)
        temp = (
            ctx.get("temp") if isinstance(ctx, dict) else getattr(ctx, "temp", None)
        )
        if isinstance(temp, dict):
            temp.setdefault("_paired_cache", {})[alias] = pair.stored
        return pair.raw

    def store(raw, ctx):
        temp = getattr(ctx, "temp", None)
        if isinstance(temp, dict):
            cached = temp.get("_paired_cache", {}).pop(alias, None)
            if cached is not None:
                return cached
        return make(ctx).stored

    cfg = _PairedCfg(
        gen=gen,
        store=store,
        alias=alias,
        verbs=tuple(verbs),
        emit=emit,
        alias_field=alias_field,
        mask_last=mask_last,
    )
    return replace(self, _paired=cfg)

alias_readtime

alias_readtime(
    name,
    fn,
    *,
    verbs=("read", "list"),
    alias_field=FieldSpec(py_type=str),
    mask_last=None,
)

Return a new spec with an additional read-time alias.

Source code in tigrbl/column/io_spec.py

def alias_readtime(
    self,
    name,
    fn,
    *,
    verbs=("read", "list"),
    alias_field: F = F(py_type=str),
    mask_last: int | None = None,
):
    """Return a new spec with an additional read-time alias."""

    alias_cfg = _ReadtimeAlias(
        name=name,
        fn=fn,
        verbs=tuple(verbs),
        alias_field=alias_field,
        mask_last=mask_last,
    )
    return replace(self, _readtime_aliases=self._readtime_aliases + (alias_cfg,))