pygx.contextual

Context-scoped value override for PyGX.

contextual

Context-scoped value override for PyGX.

pygx.contextual provides a context-local (contextvars) stack of variable bindings that inner code can read via pg.contextual.get. Bindings are installed by pg.contextual.override (a context manager); concurrent asyncio tasks each see their own bindings, context-copying mechanisms (asyncio.to_thread, anyio portals) carry them automatically, and a plain threading.Thread receives them when the callable is wrapped with pg.contextual.with_override. The module sits between pg.topology (for MISSING_VALUE) and the heavier pg.symbolic layer.

The symbols below are intentionally not re-exported at the top of pg.* — access them as pg.contextual.<name>.

Override dataclass

Override(value: Any, cascade: bool = False, override_attrs: bool = False)

Value marker for a contextual override of an attribute.

Attributes:

Name	Type	Description
value	Any	The overriding value.
cascade	bool	If True, this override applies to both the current scope and nested scopes, meaning it takes precedence over all nested pg.contextual.override bindings of the same attribute.
override_attrs	bool	If True, this override also applies to attributes that already have values.

Placeholder

Placeholder(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: ValueFromParentChain, Extension

Value marker for a contextual attribute on a ContextualObject.

Used as the default value of a field on a pg.ContextualObject subclass. At access time, the attribute resolves by walking the chain of containing ContextualObjects, then consulting any pg.contextual.override bindings in scope, and finally the marker's own default (if given); if none applies, reading the attribute raises AttributeError.

Example:

class A(pg.ContextualObject):
  y: Any = pg.contextual.Placeholder()
  z: Any = pg.contextual.Placeholder(default=0)

class B(pg.ContextualObject):
  y: int
  a: A

a = A()
assert a.z == 0        # Falls back to the marker's default.
b = B(y=2, a=a)
assert a.y == 2        # Resolved from the containing B.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

all

all() -> dict[str, Any]

Returns all values provided from pg.contextual.override in scope.

Source code in pygx/contextual/_contextual.py

def all() -> dict[str, Any]:  # pylint: disable=redefined-builtin
    """Returns all values provided from ``pg.contextual.override`` in scope."""
    overrides = _global_contextual_overrides.get({})
    return {k: v.value for k, v in overrides.items()}

current_overrides

current_overrides() -> dict[str, Override]

Returns the Override markers in scope.

A snapshot of the merged override dict the current context sees — marker objects included (unlike all(), which unwraps to plain values and drops cascade / override_attrs). Re-installing the snapshot elsewhere is spelled override(**current_overrides()); with_override and pg.concurrent.iter_sync both capture through this.

Source code in pygx/contextual/_contextual.py

def current_overrides() -> dict[str, Override]:
    """Returns the [`Override`][pygx.contextual.Override] markers in scope.

    A snapshot of the merged override dict the current context sees —
    marker objects included (unlike ``all()``, which unwraps to plain
    values and drops ``cascade`` / ``override_attrs``). Re-installing the
    snapshot elsewhere is spelled ``override(**current_overrides())``;
    `with_override` and ``pg.concurrent.iter_sync`` both capture through
    this.
    """
    return dict(_global_contextual_overrides.get({}))

get

get(var_name: str, default: Any = RAISE_IF_HAS_ERROR) -> Any

Returns the value of a variable defined in pg.contextual.override.

Parameters:

Name	Type	Description	Default
var_name	str	Name of the variable to look up in the current context's override scope.	required
default	Any	Value to return when the variable is not in scope. If omitted, a missing variable raises KeyError instead.	RAISE_IF_HAS_ERROR

Returns:

Type	Description
Any	The overridden value, or default when given and no override is in
Any	scope.

Raises:

Type	Description
KeyError	The variable is not in scope and no default was given.

Source code in pygx/contextual/_contextual.py

def get(var_name: str, default: Any = RAISE_IF_HAS_ERROR) -> Any:
    """Returns the value of a variable defined in ``pg.contextual.override``.

    Args:
      var_name: Name of the variable to look up in the current context's
        override scope.
      default: Value to return when the variable is not in scope. If omitted,
        a missing variable raises ``KeyError`` instead.

    Returns:
      The overridden value, or ``default`` when given and no override is in
      scope.

    Raises:
      KeyError: The variable is not in scope and no ``default`` was given.
    """
    o = get_override(var_name)
    if o is None:
        # Identity, not `==` — a user default that happens to compare equal
        # to the sentinel (e.g. `(pg.MISSING_VALUE,)`) must be returned, not
        # spuriously raise (#768).
        if default is RAISE_IF_HAS_ERROR:
            raise KeyError(f'{var_name!r} does not exist in current context.')
        return default
    return o.value

get_override

get_override(var_name: str) -> Override | None

Returns the in-scope Override marker.

Looks up var_name in the current context's override scope; returns None when the variable is not overridden.

Source code in pygx/contextual/_contextual.py

def get_override(var_name: str) -> Override | None:
    """Returns the in-scope [`Override`][pygx.contextual.Override] marker.

    Looks up ``var_name`` in the current context's override scope; returns
    None when the variable is not overridden.
    """
    return get_scoped_value(_global_contextual_overrides, var_name)

override

override(
    *, cascade: bool = False, override_attrs: bool = False, **variables: Any
) -> ContextManager[dict[str, Override]]

Context manager to provide contextual values under a scope.

Contextual value overrides are per-context (contextvars): concurrent asyncio tasks each see their own scopes, and mechanisms that copy the context into a worker (asyncio.to_thread, anyio portals) carry them automatically. A plain threading.Thread starts with a fresh context — wrap the user function with pg.contextual.with_override(func) to propagate the current overrides there.

Known limitation (inherent to the contextvars idiom, shared with e.g. asyncio.timeout): the scope must exit in the same contextvars.Context it entered. An async generator holding an open scope across a yield that is finalized by the event loop's shutdown_asyncgens exits in a different context and raises ValueError from the token reset. (pg.concurrent.iter_sync is not subject to this: it drives the whole iteration — every step and the close — in one task, #837.)

Example:

from typing import Any
import pygx as pg

class A(pg.ContextualObject):
  x: int
  y: Any = pg.contextual.Placeholder()

a = A(x=1)
with pg.contextual.override(y=2):
  assert a.y == 2          # `y` has no value; the override supplies it.
  assert a.x == 1          # `x` has a value; not affected by default.
with pg.contextual.override(x=3, override_attrs=True):
  assert a.x == 3          # Overrides even attributes with values.

Parameters:

Name	Type	Description	Default
cascade	bool	If True, this override applies to both the current scope and nested scopes, taking precedence over all nested overrides for the listed variables.	False
override_attrs	bool	If True, the override also applies to attributes that already have explicit values. Otherwise overridden variables are only consulted for contextual attributes whose values are not present.	False
**variables	Any	Key/values as overrides for contextual attributes.	{}

Returns:

Type	Description
ContextManager[dict[str, Override]]	A context manager that installs the overrides for its scope, yielding
ContextManager[dict[str, Override]]	a dict of attribute names to their Override
ContextManager[dict[str, Override]]	markers.

Source code in pygx/contextual/_contextual.py

def override(
    *,
    cascade: bool = False,
    override_attrs: bool = False,
    **variables: Any,
) -> ContextManager[dict[str, Override]]:
    """Context manager to provide contextual values under a scope.

    Contextual value overrides are per-context (``contextvars``): concurrent
    ``asyncio`` tasks each see their own scopes, and mechanisms that copy the
    context into a worker (``asyncio.to_thread``, anyio portals) carry them
    automatically. A plain ``threading.Thread`` starts with a fresh context —
    wrap the user function with ``pg.contextual.with_override(func)`` to
    propagate the current overrides there.

    Known limitation (inherent to the ``contextvars`` idiom, shared with
    e.g. ``asyncio.timeout``): the scope must exit in the same
    ``contextvars.Context`` it entered. An async generator holding an open
    scope across a ``yield`` that is finalized by the event loop's
    ``shutdown_asyncgens`` exits in a different context and raises
    ``ValueError`` from the token reset.
    (``pg.concurrent.iter_sync`` is not subject to this: it drives the whole
    iteration — every step and the close — in one task, #837.)

    **Example:**

    ```python
    from typing import Any
    import pygx as pg

    class A(pg.ContextualObject):
      x: int
      y: Any = pg.contextual.Placeholder()

    a = A(x=1)
    with pg.contextual.override(y=2):
      assert a.y == 2          # `y` has no value; the override supplies it.
      assert a.x == 1          # `x` has a value; not affected by default.
    with pg.contextual.override(x=3, override_attrs=True):
      assert a.x == 3          # Overrides even attributes with values.
    ```

    Args:
      cascade: If True, this override applies to both the current scope and
        nested scopes, taking precedence over all nested overrides for the
        listed variables.
      override_attrs: If True, the override also applies to attributes that
        already have explicit values. Otherwise overridden variables are only
        consulted for contextual attributes whose values are not present.
      **variables: Key/values as overrides for contextual attributes.

    Returns:
      A context manager that installs the overrides for its scope, yielding
      a dict of attribute names to their [`Override`][pygx.contextual.Override]
      markers.
    """
    vs = {}
    for k, v in variables.items():
        if not isinstance(v, Override):
            v = Override(
                value=v, cascade=cascade, override_attrs=override_attrs
            )
        vs[k] = v
    return contextual_scope(_global_contextual_overrides, **vs)

with_override

with_override(func: Callable[..., Any]) -> Callable[..., Any]

Wraps a user function so it carries the current contextual overrides.

The wrapped function may be called from another thread; the overrides active at wrap-time will be re-installed for the duration of each call. Coroutine functions are wrapped with an async wrapper. (Needed only where the contextvars.Context is NOT already carried over: a plain threading.Thread or an executor's worker thread. asyncio tasks and asyncio.to_thread copy the context by themselves.)

Example:

import threading
from typing import Any
import pygx as pg

class A(pg.ContextualObject):
  y: Any = pg.contextual.Placeholder()

a, seen = A(), []

def read(obj):
  seen.append(obj.y)

with pg.contextual.override(y=3):
  # A plain thread starts with a fresh context, so `read` must be
  # wrapped to see the overrides there.
  t = threading.Thread(target=pg.contextual.with_override(read), args=(a,))
  t.start()
  t.join()
assert seen == [3]

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	The user function to be wrapped.	required

Returns:

Type	Description
Callable[..., Any]	A wrapper function that re-installs the captured overrides on each
Callable[..., Any]	invocation, suitable for handing off to another thread.

Source code in pygx/contextual/_contextual.py

def with_override(func: Callable[..., Any]) -> Callable[..., Any]:
    """Wraps a user function so it carries the current contextual overrides.

    The wrapped function may be called from another thread; the overrides
    active at wrap-time will be re-installed for the duration of each call.
    Coroutine functions are wrapped with an async wrapper. (Needed only where
    the ``contextvars.Context`` is NOT already carried over: a plain
    ``threading.Thread`` or an executor's worker thread. ``asyncio`` tasks and
    ``asyncio.to_thread`` copy the context by themselves.)

    **Example:**

    ```python
    import threading
    from typing import Any
    import pygx as pg

    class A(pg.ContextualObject):
      y: Any = pg.contextual.Placeholder()

    a, seen = A(), []

    def read(obj):
      seen.append(obj.y)

    with pg.contextual.override(y=3):
      # A plain thread starts with a fresh context, so `read` must be
      # wrapped to see the overrides there.
      t = threading.Thread(target=pg.contextual.with_override(read), args=(a,))
      t.start()
      t.join()
    assert seen == [3]
    ```

    Args:
      func: The user function to be wrapped.

    Returns:
      A wrapper function that re-installs the captured overrides on each
      invocation, suitable for handing off to another thread.
    """
    current_context = current_overrides()

    if inspect.iscoroutinefunction(func):

        async def _async_func(*args, **kwargs) -> Any:
            with override(**current_context):  # pyright: ignore[reportArgumentType]
                return await func(*args, **kwargs)

        return _async_func

    def _func(*args, **kwargs) -> Any:
        with override(**current_context):  # pyright: ignore[reportArgumentType]
            return func(*args, **kwargs)

    return _func

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2