Class tigrbl.session.spec.SessionSpec

tigrbl.session.spec.SessionSpec dataclass

SessionSpec(
    isolation=None,
    read_only=None,
    autobegin=True,
    expire_on_commit=None,
    retry_on_conflict=None,
    max_retries=0,
    backoff_ms=0,
    backoff_jitter=True,
    statement_timeout_ms=None,
    lock_timeout_ms=None,
    fetch_rows=None,
    stream_chunk_rows=None,
    min_lsn=None,
    as_of_ts=None,
    consistency=None,
    staleness_ms=None,
    tenant_id=None,
    role=None,
    rls_context=None,
    trace_id=None,
    query_tag=None,
    tag=None,
    tracing_sample=None,
    cache_read=None,
    cache_write=None,
    namespace=None,
    kms_key_alias=None,
    classification=None,
    audit=None,
    idempotency_key=None,
    page_snapshot=None,
)

Per-session policy for Tigrbl sessions.

These fields are backend-agnostic hints and constraints. Adapters should validate and apply them where supported; critical ones (like isolation and read_only) SHOULD be validated and enforced.

isolation class-attribute instance-attribute

isolation = None

read_only class-attribute instance-attribute

read_only = None

autobegin class-attribute instance-attribute

autobegin = True

expire_on_commit class-attribute instance-attribute

expire_on_commit = None

retry_on_conflict class-attribute instance-attribute

retry_on_conflict = None

max_retries class-attribute instance-attribute

max_retries = 0

backoff_ms class-attribute instance-attribute

backoff_ms = 0

backoff_jitter class-attribute instance-attribute

backoff_jitter = True

statement_timeout_ms class-attribute instance-attribute

statement_timeout_ms = None

lock_timeout_ms class-attribute instance-attribute

lock_timeout_ms = None

fetch_rows class-attribute instance-attribute

fetch_rows = None

stream_chunk_rows class-attribute instance-attribute

stream_chunk_rows = None

min_lsn class-attribute instance-attribute

min_lsn = None

as_of_ts class-attribute instance-attribute

as_of_ts = None

consistency class-attribute instance-attribute

consistency = None

staleness_ms class-attribute instance-attribute

staleness_ms = None

tenant_id class-attribute instance-attribute

tenant_id = None

role class-attribute instance-attribute

role = None

rls_context class-attribute instance-attribute

rls_context = None

trace_id class-attribute instance-attribute

trace_id = None

query_tag class-attribute instance-attribute

query_tag = None

tag class-attribute instance-attribute

tag = None

tracing_sample class-attribute instance-attribute

tracing_sample = None

cache_read class-attribute instance-attribute

cache_read = None

cache_write class-attribute instance-attribute

cache_write = None

namespace class-attribute instance-attribute

namespace = None

kms_key_alias class-attribute instance-attribute

kms_key_alias = None

classification class-attribute instance-attribute

classification = None

audit class-attribute instance-attribute

audit = None

idempotency_key class-attribute instance-attribute

idempotency_key = None

page_snapshot class-attribute instance-attribute

page_snapshot = None

merge

merge(higher)

Overlay another spec on top of this one (non-None fields take precedence). Use to implement op > model > api > app precedence.

Source code in tigrbl/session/spec.py

def merge(self, higher: "SessionSpec | Mapping[str, Any] | None") -> "SessionSpec":
    """
    Overlay another spec on top of this one (non-None fields take precedence).
    Use to implement op > model > api > app precedence.
    """
    if higher is None:
        return self
    h = higher if isinstance(higher, SessionSpec) else SessionSpec.from_any(higher)
    if h is None:
        return self
    vals: MutableMapping[str, Any] = {f.name: getattr(self, f.name) for f in fields(SessionSpec)}
    for f in fields(SessionSpec):
        hv = getattr(h, f.name)
        if hv is not None:
            vals[f.name] = hv
    return SessionSpec(**vals)  # type: ignore[arg-type]

to_kwargs

to_kwargs()

Return only non-None items as a plain dict (adapters can **kwargs this).

Source code in tigrbl/session/spec.py

def to_kwargs(self) -> dict[str, Any]:
    """Return only non-None items as a plain dict (adapters can **kwargs this)."""
    return {f.name: getattr(self, f.name) for f in fields(SessionSpec) if getattr(self, f.name) is not None}

from_any staticmethod

from_any(x)

Source code in tigrbl/session/spec.py

@staticmethod
def from_any(x: SessionCfg) -> Optional["SessionSpec"]:
    if x is None:
        return None
    if isinstance(x, SessionSpec):
        return x
    if isinstance(x, Mapping):
        m = dict(x)
        # aliases
        if "readonly" in m and "read_only" not in m:
            m["read_only"] = bool(m.pop("readonly"))
        if "iso" in m and "isolation" not in m:
            m["isolation"] = str(m.pop("iso"))
        allowed = {f.name for f in fields(SessionSpec)}
        return SessionSpec(**{k: v for k, v in m.items() if k in allowed})
    raise TypeError(f"Unsupported SessionSpec input: {type(x)}")