pygx.contextual

Context-scoped value override for PyGX.

contextual

Context-scoped value override for PyGX.

pygx.contextual provides a thread-local stack of variable bindings that inner code can read via pg.contextual.get. Bindings are installed by pg.contextual.override (a context manager) and propagate across threads when a 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.

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 containing object chain and consulting any pg.contextual.override bindings in scope.

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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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 = getattr(
        _global_contextual_overrides, _TLS_KEY_CONTEXTUAL_OVERRIDES, {}
    )
    return {k: v.value for k, v in overrides.items()}

get

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

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

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`."""
    o = get_override(var_name)
    if o is None:
        if default == 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 overridden contextual marker for var_name in scope.

Source code in pygx/contextual/_contextual.py

def get_override(var_name: str) -> Override | None:
    """Returns the overridden contextual marker for ``var_name`` in scope."""
    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-thread. To propagate them to other threads, wrap the user function with pg.contextual.with_override(func).

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 dict of attribute names to their 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-thread. To propagate them to other
    threads, wrap the user function with `pg.contextual.with_override(func)`.

    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 dict of attribute names to their `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.

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.

    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.
    """
    with override() as current_context:
        pass

    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