Class tigrbl.runtime.context.Context

tigrbl.runtime.context.Context dataclass

Context(
    op,
    persist,
    specs,
    cfg,
    temp=dict(),
    model=None,
    obj=None,
    session=None,
    user=None,
    tenant=None,
    now=None,
    row=None,
    values=None,
    current_values=None,
    in_data=None,
    payload=None,
    data=None,
    body=None,
)

Canonical runtime context shared by the kernel and atoms.

Minimal contract (consumed by atoms we’ve written so far): - op: operation name (e.g., 'create' | 'update' | 'read' | 'list' | custom) - persist: write vs. read (affects pruning of persist-tied anchors) - specs: mapping of field -> ColumnSpec (frozen at bind time) - cfg: read-only config view (see config.resolver.CfgView) - temp: dict scratchpad used by atoms to exchange data

Optional adapter slots

• model: owning model type / class
• obj: hydrated ORM instance (if any)
• session: DB session / unit-of-work handle
• user, tenant, now: identity/time hints
• row/values/current_values: mapping fallbacks (for read paths)
• in_data / payload / data / body: inbound payload staging (for build_in)

op instance-attribute

op

persist instance-attribute

persist

specs instance-attribute

specs

cfg instance-attribute

cfg

temp class-attribute instance-attribute

temp = field(default_factory=dict)

model class-attribute instance-attribute

model = None

obj class-attribute instance-attribute

obj = None

session class-attribute instance-attribute

session = None

user class-attribute instance-attribute

user = None

tenant class-attribute instance-attribute

tenant = None

now class-attribute instance-attribute

now = None

row class-attribute instance-attribute

row = None

values class-attribute instance-attribute

values = None

current_values class-attribute instance-attribute

current_values = None

in_data class-attribute instance-attribute

in_data = None

payload class-attribute instance-attribute

payload = None

data class-attribute instance-attribute

data = None

body class-attribute instance-attribute

body = None

is_write property

is_write

Alias for persist; reads better in some call sites.

safe_view

safe_view(*, include_temp=False, temp_keys=None)

Return a small, read-only mapping exposing only safe, frequently useful keys.

By default, temp is NOT included (to avoid leaking internals like paired raw values). If include_temp=True, only exposes the keys listed in 'temp_keys' (if provided), otherwise exposes a conservative subset.

This method is intended to be passed into author callables such as default_factory(ctx_view) or paired token generators.

Source code in tigrbl/runtime/context.py

def safe_view(
    self,
    *,
    include_temp: bool = False,
    temp_keys: Optional[Sequence[str]] = None,
) -> Mapping[str, Any]:
    """
    Return a small, read-only mapping exposing only safe, frequently useful keys.

    By default, temp is NOT included (to avoid leaking internals like paired raw values).
    If include_temp=True, only exposes the keys listed in 'temp_keys' (if provided),
    otherwise exposes a conservative subset.

    This method is intended to be passed into author callables such as
    default_factory(ctx_view) or paired token generators.
    """
    base = {
        "op": self.op,
        "persist": self.persist,
        "model": self.model,
        "specs": self.specs,
        "user": self.user,
        "tenant": self.tenant,
        "now": self.now,
    }
    if include_temp:
        allowed = set(temp_keys or ("assembled_values", "virtual_in"))
        exposed: Dict[str, Any] = {}
        for k in allowed:
            if k in self.temp:
                exposed[k] = self.temp[k]
        base = {**base, "temp": MappingProxy(exposed)}
    return MappingProxy(base)

mark_used_returning

mark_used_returning(value=True)

Flag that DB RETURNING already hydrated values.

Source code in tigrbl/runtime/context.py

def mark_used_returning(self, value: bool = True) -> None:
    """Flag that DB RETURNING already hydrated values."""
    self.temp["used_returning"] = bool(value)

merge_hydrated_values

merge_hydrated_values(mapping, *, replace=False)

Save values hydrated from DB (RETURNING/refresh). If replace=False (default), performs a shallow merge into any existing 'hydrated_values'.

Source code in tigrbl/runtime/context.py

def merge_hydrated_values(
    self, mapping: Mapping[str, Any], *, replace: bool = False
) -> None:
    """
    Save values hydrated from DB (RETURNING/refresh). If replace=False (default),
    performs a shallow merge into any existing 'hydrated_values'.
    """
    if not isinstance(mapping, Mapping):
        return
    hv = self.temp.get("hydrated_values")
    if replace or not isinstance(hv, dict):
        self.temp["hydrated_values"] = dict(mapping)
    else:
        hv.update(mapping)

add_response_extras

add_response_extras(extras, *, overwrite=None)

Merge alias extras into temp['response_extras']. Returns a tuple of conflicting keys that were skipped when overwrite=False.

Source code in tigrbl/runtime/context.py

def add_response_extras(
    self, extras: Mapping[str, Any], *, overwrite: Optional[bool] = None
) -> Sequence[str]:
    """
    Merge alias extras into temp['response_extras'].
    Returns a tuple of conflicting keys that were skipped when overwrite=False.
    """
    if not isinstance(extras, Mapping) or not extras:
        return ()
    buf = self.temp.get("response_extras")
    if not isinstance(buf, dict):
        buf = {}
        self.temp["response_extras"] = buf
    if overwrite is None:
        # fall back to cfg; atoms call wire:dump to honor final overwrite policy
        overwrite = bool(getattr(self.cfg, "response_extras_overwrite", False))
    conflicts: list[str] = []
    for k, v in extras.items():
        if (k in buf) and not overwrite:
            conflicts.append(k)
            continue
        buf[k] = v
    return tuple(conflicts)

get_response_payload

get_response_payload()

Return the payload assembled by wire:dump (or None if not yet available).

Source code in tigrbl/runtime/context.py

def get_response_payload(self) -> Any:
    """Return the payload assembled by wire:dump (or None if not yet available)."""
    return self.temp.get("response_payload")