pygx.patching

Systematic patching on symbolic values.

patching

Systematic patching on symbolic values.

While pygx.Symbolic.sym_rebind already provides a flexible programming interface for modifying symbolic values, this module builds on it to:

• Provide user-friendly helpers for the most common patching patterns, such as patching by key, path, type, value or class member.

• Package related changes into reusable, composable semantic groups (pygx.patching.Patcher), which can also be created from URI-like strings so patching can be driven from the command line.

Patcher

Patcher(
    *args: Any,
    root_path: KeyPath | None = None,
    override_args: bool = False,
    ignore_extra_args: bool = False,
    **kwargs: Any
)

Bases: Functor

Class that patches a symbolic value by returning a rebind dict.

To accommodate reusable patching, we introduced the concept of patcher, which is a symbolic function that takes an input value with a list of optional arguments and produces a patching rule (see pygx.patch) or a tuple of (patching_rule, validation_rule). A validation rule is a callable object that validates the patched object's integrity if it's being patched by others later. A patcher can be created from a URI-like string, to better serve the command-line interface.

A patcher can be created via the pg.patcher decorator, returning a rebind dict:

@pg.patcher([
    ('lr', pg.typing.Float(min_value=0.0))
])
def learning_rate(trainer, lr):
  return {
      'training.learning_rate': lr
  }

or a rebind function:

@pg.patcher([
    ('lr', pg.typing.Float(min_value=0.0))
])
def learning_rate(trainer, lr):
  def _rebind_fn(k, v, p):
    if k and k.key == 'learning_rate' and isinstance(v, float):
      return lr
    return v
  return _rebind_fn

or a composition of rules:

@pg.patcher([
    ('lr', pg.typing.Float(min_value=0.0)),
    ('weight_decay', pg.typing.Float(min_value=0.0))
])
def complex(trainer, lr, weight_decay):
  # `change_lr` and `change_weight_decay` are other patcher classes.
  # Their arguments must be passed by keyword: the first positional
  # argument of a patcher is reserved for the patching target.
  return [
      change_lr(lr=lr),
      change_weight_decay(weight_decay=weight_decay)
  ]

After registration, a patcher object can be obtained from a URI-like string:

patcher = pg.patching.from_uri('cosine_decay?lr=0.1')

Then the user can use the patcher to patch an object:

patched_trainer = patcher.patch(trainer)

A list of patchers can be applied sequentially to accommodate combination of semantic groups. A patcher in the sequence can propose updates to the original value or generate a replacement for the original value. pg.patching.patch is introduced for making it convenient to chain multiple patchers using URI-like strings:

pg.patching.patch(trainer, [
    'cosine_decay?lr=0.1',
    'some_other_patcher_string',
])

The user can look up all registered patchers via:

print(pg.patching.patcher_names())

Source code in pygx/symbolic/_functor.py

def __init__(
    self,
    *args: Any,
    root_path: topology.KeyPath | None = None,
    override_args: bool = False,
    ignore_extra_args: bool = False,
    **kwargs: Any,
):
    """Constructor.

    Args:
      *args: prebound positional arguments.
      root_path: The symbolic path for current object.
      override_args: If True, allows arguments provided during `__call__` to
        override existing bound arguments.
      ignore_extra_args: If True, unsupported arguments can be passed in
        during `__call__` without using them. Otherwise, calling with
        unsupported arguments will raise error.
      **kwargs: prebound keyword arguments.

    Raises:
      TypeError: constructor got unexpected or duplicate arguments.
    """
    # NOTE(daiyip): Since Functor is usually late bound (until call time),
    # we pass `allow_partial=True` during functor construction.
    _ = kwargs.pop('allow_partial', None)

    varargs = None
    signature = self.__signature__
    if len(args) > len(signature.args):
        if signature.varargs:
            varargs = list(args[len(signature.args) :])
            args = args[: len(signature.args)]
        else:
            arg_phrase = formatting.auto_plural(
                len(signature.args), 'argument'
            )
            was_phrase = formatting.auto_plural(len(args), 'was', 'were')
            raise TypeError(
                f'{signature.id}() takes {len(signature.args)} '
                f'positional {arg_phrase} but {len(args)} {was_phrase} given.'
            )

    bound_kwargs = dict()
    for i, v in enumerate(args):
        if pg_typing.MISSING_VALUE != v:
            bound_kwargs[signature.args[i].name] = v

    if varargs is not None:
        bound_kwargs[signature.varargs.name] = varargs

    for k, v in kwargs.items():
        if pg_typing.MISSING_VALUE != v:
            if k in bound_kwargs:
                raise TypeError(
                    f'{signature.id}() got multiple values for keyword '
                    f'argument {k!r}.'
                )
            bound_kwargs[k] = v

    default_args = set()
    non_default_args = set(bound_kwargs)

    for arg_spec in signature.named_args:
        if not arg_spec.value_spec.has_default:
            continue
        arg_name = arg_spec.name
        if arg_name not in non_default_args:
            default_args.add(arg_name)
        elif bound_kwargs[arg_name] == arg_spec.value_spec.default:
            default_args.add(arg_name)
            non_default_args.discard(arg_name)

    if signature.varargs and not varargs:
        default_args.add(signature.varargs.name)

    super().__init__(
        allow_partial=True, root_path=root_path, **bound_kwargs
    )

    self._non_default_args = non_default_args
    self._default_args = default_args
    self._specified_args = set(bound_kwargs)
    self._override_args = override_args
    self._ignore_extra_args = ignore_extra_args

    # For subclassed Functor, we use thread-local storage for storing temporary
    # member overrides from the arguments during functor call.
    self._tls = threading.local() if self.is_subclassed_functor() else None

patch

patch(x: Symbolic) -> Any

Patches an input and returns the input itself unless replaced.

Calls the patcher body with x to obtain the patching rule (and optional validation rule), then applies the rule via pygx.patch. Returns a different object only when a rule replaces the whole value.

Source code in pygx/patching/_rule_based.py

def patch(self, x: symbolic.Symbolic) -> Any:
    """Patches an input and returns the input itself unless replaced.

    Calls the patcher body with ``x`` to obtain the patching rule (and
    optional validation rule), then applies the rule via
    [`pygx.patch`][pygx.patching.patch]. Returns a different object only
    when a rule replaces the whole value.
    """
    # Placeholder for Google-internal usage instrumentation.

    if not isinstance(x, symbolic.Symbolic):
        raise TypeError(
            f'The 1st argument of {self.__class__.__name__!r} must be a '
            f'symbolic type. Encountered: {x!r}.'
        )
    # The patching rule returned from the patcher body should be either
    # (patching_rule, validation_rule) or just patching_rule.
    # The patching rule can be a dict, a rebind function, another patcher
    # or a mixture of them as a list. The patching rule is then passed to
    # ``pg.patch`` which allows patchers to support composition of sub-patchers.
    patching_rule = self(x)

    # Set validator if applicable.
    validator = None
    if isinstance(patching_rule, tuple) and len(patching_rule) == 2:
        patching_rule, validator = patching_rule
    if validator is not None and not callable(validator):
        raise TypeError(
            f'The validator returned from patcher {self.__class__.__name__!r} '
            f'is not callable. Encountered: {validator!r}.'
        )
    self._validator = validator
    return patch(x, patching_rule)

validate

validate(x: Symbolic) -> None

Validates an input's integrity.

This method will be called in pygx.patch after a chain of patchers has been applied, to validate that the patched object still conforms to this patcher's plan.

Parameters:

Name	Type	Description	Default
x	Symbolic	The input after modification.	required

Source code in pygx/patching/_rule_based.py

def validate(self, x: symbolic.Symbolic) -> None:
    """Validates an input's integrity.

    This method will be called in [`pygx.patch`][pygx.patching.patch]
    after a chain of patchers has been applied, to validate that the
    patched object still conforms to this patcher's plan.

    Args:
      x: The input after modification.
    """
    if self._validator is not None:
        self._validator(x)

ObjectFactory

ObjectFactory(
    value_type: type[Symbolic],
    base_value: Symbolic | Callable[[], Symbolic] | str,
    patches: PatchType | None = None,
    params_override: dict[str, Any] | str | None = None,
) -> Any

A factory to create a symbolic object from a base value and patches.

ObjectFactory is a functor: calling it with arguments returns a factory object, and calling that object produces the patched value.

Example:

class Trainer(pg.Object):
  lr: float = 0.1
  steps: int = 100

factory = pg.patching.ObjectFactory(
    Trainer, Trainer(), {'lr': 0.01})
trainer = factory()
assert trainer.lr == 0.01

Parameters:

Name	Type	Description	Default
value_type	type[Symbolic]	Type of the return value.	required
base_value	Symbolic | Callable[[], Symbolic] | str	An instance of value_type, or a callable object that produces an instance of value_type, or a string as the path to the serialized value.	required
patches	PatchType | None	Optional patching rules. See pygx.patch for details.	None
params_override	dict[str, Any] | str | None	A rebind dict (or a JSON string as a serialized rebind dict) as an additional patch to the value.	None

Returns:

Type	Description
Any	Value after applying patches and params_override based on
Any	base_value.

Source code in pygx/patching/_object_factory.py

@symbolic.functor()
def ObjectFactory(  # pylint: disable=invalid-name
    value_type: type[symbolic.Symbolic],
    base_value: symbolic.Symbolic | Callable[[], symbolic.Symbolic] | str,
    patches: rule_based.PatchType | None = None,
    params_override: dict[str, Any] | str | None = None,
) -> Any:
    """A factory to create a symbolic object from a base value and patches.

    ``ObjectFactory`` is a functor: calling it with arguments returns a
    factory object, and calling that object produces the patched value.

    **Example:**

    ```python
    class Trainer(pg.Object):
      lr: float = 0.1
      steps: int = 100

    factory = pg.patching.ObjectFactory(
        Trainer, Trainer(), {'lr': 0.01})
    trainer = factory()
    assert trainer.lr == 0.01
    ```

    Args:
      value_type: Type of the return value.
      base_value: An instance of ``value_type``, or a callable object that
        produces an instance of ``value_type``, or a string as the path to
        the serialized value.
      patches: Optional patching rules. See
        [`pygx.patch`][pygx.patching.patch] for details.
      params_override: A rebind dict (or a JSON string as a serialized rebind
        dict) as an additional patch to the value.

    Returns:
      Value after applying ``patches`` and ``params_override`` based on
      ``base_value``.
    """
    # Step 1: Load base value.
    if not isinstance(base_value, value_type) and callable(base_value):
        value = base_value()
    elif isinstance(base_value, str):
        value = symbolic.load(base_value)
    else:
        value = base_value

    if not isinstance(value, value_type):
        raise TypeError(
            f'{base_value!r} is neither an instance of {value_type!r}, '
            f'nor a factory or a path of JSON file that produces an '
            f'instance of {value_type!r}.'
        )

    # Step 2: Patch with patchers if available.
    if patches is not None:
        value = rule_based.patch(value, patches)

    # Step 3: Patch with additional parameter override dict if available.
    if params_override:
        value = value.sym_rebind(
            topology.flatten(from_maybe_serialized(params_override, dict)),
        )
    return value

from_maybe_serialized

from_maybe_serialized(
    source: Any | str, value_type: type[Any] | None = None
) -> Any

Loads a value from a maybe-serialized form (JSON file or JSON string).

Parameters:

Name	Type	Description	Default
source	Any | str	Source of the value. It can be the value (non-string type) itself, or a file path (a string ending with .json), or a JSON string from which the value will be loaded.	required
value_type	type[Any] | None	An optional type to constrain the value.	None

Returns:

Type	Description
Any	Value from source.

Raises:

Type	Description
TypeError	If the loaded value is not an instance of value_type.

Source code in pygx/patching/_object_factory.py

def from_maybe_serialized(
    source: Any | str, value_type: type[Any] | None = None
) -> Any:
    """Loads a value from a maybe-serialized form (JSON file or JSON string).

    Args:
      source: Source of the value. It can be the value (non-string type)
        itself, or a file path (a string ending with ``.json``), or a JSON
        string from which the value will be loaded.
      value_type: An optional type to constrain the value.

    Returns:
      Value from source.

    Raises:
      TypeError: If the loaded value is not an instance of ``value_type``.
    """
    if isinstance(source, str):
        if source.endswith('.json'):
            value = symbolic.load(source)
        else:
            value = symbolic.from_json_str(source)
    else:
        value = source
    if value_type is not None and not isinstance(value, value_type):
        raise TypeError(
            f'Loaded value {value!r} is not an instance of {value_type!r}.'
        )
    return value

patch_on_key

patch_on_key(
    src: Symbolic,
    regex: str,
    value: Any = MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values whose keys match a regular expression.

Every node in the symbolic tree is tested - containers included - so a matching key replaces the entire value stored under it. The key is the last component of a node's path (e.g. 'b' for path 'a.b').

Example:

d = pg.Dict(a=0, b=2)
pg.patching.patch_on_key(d, 'a', value=3)
# d is now {a=3, b=2}.

pg.patching.patch_on_key(d, '.', value=4)
# d is now {a=4, b=4}.

class A(pg.Object):
  x: int

  def on_sym_post_init(self):
    super().on_sym_post_init()
    self._num_changes = 0

  def on_sym_change(self, updates):
    super().on_sym_change(updates)
    self._num_changes += 1

a = A(x=1)
pg.patching.patch_on_key(a, 'x', value=2)
# a._num_changes is 1.

pg.patching.patch_on_key(a, 'x', value=3)
# a._num_changes is 2.

pg.patching.patch_on_key(a, 'x', value=4, skip_notification=True)
# a._num_changes is still 2.

Parameters:

Name	Type	Description	Default
src	Symbolic	Symbolic value to patch.	required
regex	str	Regex for the key name, matched from the beginning of the name (via re.match).	required
value	Any	New value for the fields whose keys match regex. An explicit None patches the field TO None (the unset default is an internal sentinel, not None). Mutually exclusive with value_fn; one of the two must be provided.	MISSING_VALUE
value_fn	Callable[[Any], Any] | None	Callable object that produces a new value based on the old value. Mutually exclusive with value.	None
skip_notification	bool | None	If True, the on_sym_change event will not be triggered for this operation. If None, the default is inferred from the pg.notify_on_change context manager. See pygx.Symbolic.sym_rebind for details.	None

Returns:

Type	Description
Any	src after being patched.

Source code in pygx/patching/_pattern_based.py

def patch_on_key(
    src: symbolic.Symbolic,
    regex: str,
    value: Any = topology.MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values whose keys match a regular expression.

    Every node in the symbolic tree is tested - containers included - so a
    matching key replaces the entire value stored under it. The key is the
    last component of a node's path (e.g. ``'b'`` for path ``'a.b'``).

    **Example:**

    ```python
    d = pg.Dict(a=0, b=2)
    pg.patching.patch_on_key(d, 'a', value=3)
    # d is now {a=3, b=2}.

    pg.patching.patch_on_key(d, '.', value=4)
    # d is now {a=4, b=4}.

    class A(pg.Object):
      x: int

      def on_sym_post_init(self):
        super().on_sym_post_init()
        self._num_changes = 0

      def on_sym_change(self, updates):
        super().on_sym_change(updates)
        self._num_changes += 1

    a = A(x=1)
    pg.patching.patch_on_key(a, 'x', value=2)
    # a._num_changes is 1.

    pg.patching.patch_on_key(a, 'x', value=3)
    # a._num_changes is 2.

    pg.patching.patch_on_key(a, 'x', value=4, skip_notification=True)
    # a._num_changes is still 2.
    ```

    Args:
      src: Symbolic value to patch.
      regex: Regex for the key name, matched from the beginning of the name
        (via ``re.match``).
      value: New value for the fields whose keys match ``regex``.
        An explicit ``None`` patches the field TO ``None`` (the unset
        default is an internal sentinel, not ``None``). Mutually exclusive
        with ``value_fn``; one of the two must be provided.
      value_fn: Callable object that produces a new value based on the old
        value. Mutually exclusive with ``value``.
      skip_notification: If True, the ``on_sym_change`` event will not be
        triggered for this operation. If None, the default is inferred from
        the [`pg.notify_on_change`][pygx.symbolic.notify_on_change] context
        manager. See
        [`pygx.Symbolic.sym_rebind`][pygx.symbolic.Symbolic.sym_rebind] for
        details.

    Returns:
      ``src`` after being patched.
    """
    compiled_regex = re.compile(regex)
    return _conditional_patch(
        src,
        lambda k, v, p: bool(k and compiled_regex.match(str(k.key))),
        value,
        value_fn,
        skip_notification,
    )

patch_on_member

patch_on_member(
    src: Symbolic,
    cls: type[Any] | tuple[type[Any], ...],
    name: str,
    value: Any = MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values that are the requested member of classes.

Example:

class A(pg.Object):
  x: int

d = pg.Dict(a=A(x=1), b=2)
pg.patching.patch_on_member(d, A, 'x', 2)
# d is now {a=A(x=2), b=2}.

Parameters:

Name	Type	Description	Default
src	Symbolic	Symbolic value to patch.	required
cls	type[Any] | tuple[type[Any], ...]	The class (or classes) the member belongs to.	required
name	str	Member name.	required
value	Any	New value for the matched members. An explicit None patches the field TO None (the unset default is an internal sentinel, not None). Mutually exclusive with value_fn; one of the two must be provided.	MISSING_VALUE
value_fn	Callable[[Any], Any] | None	Callable object that produces a new value based on the old value. Mutually exclusive with value.	None
skip_notification	bool | None	If True, the on_sym_change event will not be triggered for this operation. If None, the default is inferred from the pg.notify_on_change context manager. See pygx.Symbolic.sym_rebind for details.	None

Returns:

Type	Description
Any	src after being patched.

Source code in pygx/patching/_pattern_based.py

def patch_on_member(
    src: symbolic.Symbolic,
    cls: type[Any] | tuple[type[Any], ...],
    name: str,
    value: Any = topology.MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values that are the requested member of classes.

    **Example:**

    ```python
    class A(pg.Object):
      x: int

    d = pg.Dict(a=A(x=1), b=2)
    pg.patching.patch_on_member(d, A, 'x', 2)
    # d is now {a=A(x=2), b=2}.
    ```

    Args:
      src: Symbolic value to patch.
      cls: The class (or classes) the member belongs to.
      name: Member name.
      value: New value for the matched members.
        An explicit ``None`` patches the field TO ``None`` (the unset
        default is an internal sentinel, not ``None``). Mutually exclusive
        with ``value_fn``; one of the two must be provided.
      value_fn: Callable object that produces a new value based on the old
        value. Mutually exclusive with ``value``.
      skip_notification: If True, the ``on_sym_change`` event will not be
        triggered for this operation. If None, the default is inferred from
        the [`pg.notify_on_change`][pygx.symbolic.notify_on_change] context
        manager. See
        [`pygx.Symbolic.sym_rebind`][pygx.symbolic.Symbolic.sym_rebind] for
        details.

    Returns:
      ``src`` after being patched.
    """
    return _conditional_patch(
        src,
        lambda k, v, p: isinstance(p, cls) and k.key == name,
        value,
        value_fn,
        skip_notification,
    )

patch_on_path

patch_on_path(
    src: Symbolic,
    regex: str,
    value: Any = MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values on matched paths.

Example:

d = pg.Dict(a={'x': 1}, b=2)
pg.patching.patch_on_path(d, '.*x', value=3)
# d is now {a={x=3}, b=2}.

Parameters:

Name	Type	Description	Default
src	Symbolic	Symbolic value to patch.	required
regex	str	Regex for the key path (e.g. 'a.b[0]'), matched from the beginning of the path (via re.match).	required
value	Any	New value for the fields whose paths match regex. An explicit None patches the field TO None (the unset default is an internal sentinel, not None). Mutually exclusive with value_fn; one of the two must be provided.	MISSING_VALUE
value_fn	Callable[[Any], Any] | None	Callable object that produces a new value based on the old value. Mutually exclusive with value.	None
skip_notification	bool | None	If True, the on_sym_change event will not be triggered for this operation. If None, the default is inferred from the pg.notify_on_change context manager. See pygx.Symbolic.sym_rebind for details.	None

Returns:

Type	Description
Any	src after being patched.

Source code in pygx/patching/_pattern_based.py

def patch_on_path(
    src: symbolic.Symbolic,
    regex: str,
    value: Any = topology.MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values on matched paths.

    **Example:**

    ```python
    d = pg.Dict(a={'x': 1}, b=2)
    pg.patching.patch_on_path(d, '.*x', value=3)
    # d is now {a={x=3}, b=2}.
    ```

    Args:
      src: Symbolic value to patch.
      regex: Regex for the key path (e.g. ``'a.b[0]'``), matched from the
        beginning of the path (via ``re.match``).
      value: New value for the fields whose paths match ``regex``.
        An explicit ``None`` patches the field TO ``None`` (the unset
        default is an internal sentinel, not ``None``). Mutually exclusive
        with ``value_fn``; one of the two must be provided.
      value_fn: Callable object that produces a new value based on the old
        value. Mutually exclusive with ``value``.
      skip_notification: If True, the ``on_sym_change`` event will not be
        triggered for this operation. If None, the default is inferred from
        the [`pg.notify_on_change`][pygx.symbolic.notify_on_change] context
        manager. See
        [`pygx.Symbolic.sym_rebind`][pygx.symbolic.Symbolic.sym_rebind] for
        details.

    Returns:
      ``src`` after being patched.
    """
    compiled_regex = re.compile(regex)
    return _conditional_patch(
        src,
        lambda k, v, p: bool(compiled_regex.match(str(k))),
        value,
        value_fn,
        skip_notification,
    )

patch_on_type

patch_on_type(
    src: Symbolic,
    value_type: type[Any] | tuple[type[Any], ...],
    value: Any = MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values on matched types.

Example:

d = pg.Dict(a={'x': 1}, b=2)
pg.patching.patch_on_type(d, int, value_fn=lambda x: x * 2)
# d is now {a={x=2}, b=4}.

Parameters:

Name	Type	Description	Default
src	Symbolic	Symbolic value to patch.	required
value_type	type[Any] | tuple[type[Any], ...]	Value type to match.	required
value	Any	New value for the fields whose values are instances of value_type. An explicit None patches the field TO None (the unset default is an internal sentinel, not None). Mutually exclusive with value_fn; one of the two must be provided.	MISSING_VALUE
value_fn	Callable[[Any], Any] | None	Callable object that produces a new value based on the old value. Mutually exclusive with value.	None
skip_notification	bool | None	If True, the on_sym_change event will not be triggered for this operation. If None, the default is inferred from the pg.notify_on_change context manager. See pygx.Symbolic.sym_rebind for details.	None

Returns:

Type	Description
Any	src after being patched.

Source code in pygx/patching/_pattern_based.py

def patch_on_type(
    src: symbolic.Symbolic,
    value_type: type[Any] | tuple[type[Any], ...],
    value: Any = topology.MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values on matched types.

    **Example:**

    ```python
    d = pg.Dict(a={'x': 1}, b=2)
    pg.patching.patch_on_type(d, int, value_fn=lambda x: x * 2)
    # d is now {a={x=2}, b=4}.
    ```

    Args:
      src: Symbolic value to patch.
      value_type: Value type to match.
      value: New value for the fields whose values are instances of
        ``value_type``.
        An explicit ``None`` patches the field TO ``None`` (the unset
        default is an internal sentinel, not ``None``). Mutually exclusive
        with ``value_fn``; one of the two must be provided.
      value_fn: Callable object that produces a new value based on the old
        value. Mutually exclusive with ``value``.
      skip_notification: If True, the ``on_sym_change`` event will not be
        triggered for this operation. If None, the default is inferred from
        the [`pg.notify_on_change`][pygx.symbolic.notify_on_change] context
        manager. See
        [`pygx.Symbolic.sym_rebind`][pygx.symbolic.Symbolic.sym_rebind] for
        details.

    Returns:
      ``src`` after being patched.
    """
    return _conditional_patch(
        src,
        lambda k, v, p: isinstance(v, value_type),
        value,
        value_fn,
        skip_notification,
    )

patch_on_value

patch_on_value(
    src: Symbolic,
    old_value: Any,
    value: Any = MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values on matched values.

Example:

d = pg.Dict(a={'x': 1}, b=1)
pg.patching.patch_on_value(d, 1, value=3)
# d is now {a={x=3}, b=3}.

Parameters:

Name	Type	Description	Default
src	Symbolic	Symbolic value to patch.	required
old_value	Any	Old value to match.	required
value	Any	New value for the fields whose values equal old_value. An explicit None patches the field TO None (the unset default is an internal sentinel, not None). Mutually exclusive with value_fn; one of the two must be provided.	MISSING_VALUE
value_fn	Callable[[Any], Any] | None	Callable object that produces a new value based on the old value. Mutually exclusive with value.	None
skip_notification	bool | None	If True, the on_sym_change event will not be triggered for this operation. If None, the default is inferred from the pg.notify_on_change context manager. See pygx.Symbolic.sym_rebind for details.	None

Returns:

Type	Description
Any	src after being patched.

Source code in pygx/patching/_pattern_based.py

def patch_on_value(
    src: symbolic.Symbolic,
    old_value: Any,
    value: Any = topology.MISSING_VALUE,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values on matched values.

    **Example:**

    ```python
    d = pg.Dict(a={'x': 1}, b=1)
    pg.patching.patch_on_value(d, 1, value=3)
    # d is now {a={x=3}, b=3}.
    ```

    Args:
      src: Symbolic value to patch.
      old_value: Old value to match.
      value: New value for the fields whose values equal ``old_value``.
        An explicit ``None`` patches the field TO ``None`` (the unset
        default is an internal sentinel, not ``None``). Mutually exclusive
        with ``value_fn``; one of the two must be provided.
      value_fn: Callable object that produces a new value based on the old
        value. Mutually exclusive with ``value``.
      skip_notification: If True, the ``on_sym_change`` event will not be
        triggered for this operation. If None, the default is inferred from
        the [`pg.notify_on_change`][pygx.symbolic.notify_on_change] context
        manager. See
        [`pygx.Symbolic.sym_rebind`][pygx.symbolic.Symbolic.sym_rebind] for
        details.

    Returns:
      ``src`` after being patched.
    """
    return _conditional_patch(
        src, lambda k, v, p: v == old_value, value, value_fn, skip_notification
    )

allow_repeated_patcher_registration

allow_repeated_patcher_registration(allow: bool = True)

Sets whether registering an already-registered patcher name is allowed.

Allowed by default; a repeated registration replaces the previous patcher. When disallowed, registering a duplicate name raises KeyError.

Source code in pygx/patching/_rule_based.py

def allow_repeated_patcher_registration(allow: bool = True):
    """Sets whether registering an already-registered patcher name is allowed.

    Allowed by default; a repeated registration replaces the previous
    patcher. When disallowed, registering a duplicate name raises
    ``KeyError``.
    """
    global _ALLOW_REPEATED_PATCHER_REGISTRATION  # pylint: disable=global-statement
    _ALLOW_REPEATED_PATCHER_REGISTRATION = allow

from_uri

from_uri(uri: str) -> Patcher

Creates a Patcher object from a URI-like string.

The format is <patcher_name>?<arg1>&<arg2>=<val2>: the name of a registered patcher, optionally followed by positional and keyword arguments as strings, which are parsed/validated against the patcher's value specs. The returned patcher is unbound from any target; apply it with its patch() method.

Example:

@pg.patching.patcher([('x', pg.typing.Int())])
def increment(v, x=1):
  return lambda k, v, p: v + x if isinstance(v, int) else v

p = pg.patching.from_uri('increment?x=3')
assert p.patch(pg.Dict(a=1)) == {'a': 4}

Raises:

Type	Description
KeyError	If the patcher name is not registered.
ValueError	If the URI is malformed or arguments fail to parse.

Source code in pygx/patching/_rule_based.py

def from_uri(uri: str) -> Patcher:
    """Creates a Patcher object from a URI-like string.

    The format is ``<patcher_name>?<arg1>&<arg2>=<val2>``: the name of a
    registered patcher, optionally followed by positional and keyword
    arguments as strings, which are parsed/validated against the patcher's
    value specs. The returned patcher is unbound from any target; apply it
    with its ``patch()`` method.

    **Example:**

    ```python
    @pg.patching.patcher([('x', pg.typing.Int())])
    def increment(v, x=1):
      return lambda k, v, p: v + x if isinstance(v, int) else v

    p = pg.patching.from_uri('increment?x=3')
    assert p.patch(pg.Dict(a=1)) == {'a': 4}
    ```

    Raises:
      KeyError: If the patcher name is not registered.
      ValueError: If the URI is malformed or arguments fail to parse.
    """
    name, args, kwargs = parse_uri(uri)
    patcher_cls = typing.cast(type[Any], _PATCHER_REGISTRY.get(name))
    args, kwargs = parse_args(patcher_cls.__signature__, args, kwargs)
    return patcher_cls(topology.MISSING_VALUE, *args, **kwargs)

patch

patch(value: Symbolic, rule: PatchType) -> Any

Apply patches to a symbolic value.

Parameters:

Name	Type	Description	Default
value	Symbolic	A symbolic value to patch.	required
rule	PatchType	A patching rule is one of the following: 1) A dict of symbolic paths to the new values. 2) A rebind function defined by signature (k, v) or (k, v, p). See pygx.Symbolic.sym_rebind. 3) A pygx.patching.Patcher object. 4) A URI-like string representing an instance of a registered Patcher. The format is <patcher_name>?<arg1>&<arg2>=<val2> (positional arguments before keyword arguments). 5) A list of the mixtures of the above.	required

Returns:

Type	Description
Any	Value after applying the patchers. If any patcher returned a new value
Any	(by returning a single-item dict containing '' as its key), the return
Any	value will be a different object than value; otherwise value
Any	will be returned after applying the patches.

Raises:

Type	Description
ValueError	Error if the patch name and arguments cannot be parsed successfully.

Source code in pygx/patching/_rule_based.py

def patch(value: symbolic.Symbolic, rule: PatchType) -> Any:
    """Apply patches to a symbolic value.

    Args:
      value: A symbolic value to patch.
      rule: A patching rule is one of the following:
        1) A dict of symbolic paths to the new values.
        2) A rebind function defined by signature (k, v) or (k, v, p).
           See [`pygx.Symbolic.sym_rebind`][pygx.symbolic.Symbolic.sym_rebind].
        3) A [`pygx.patching.Patcher`][pygx.patching.Patcher] object.
        4) A URI-like string representing an instance of a registered
           Patcher. The format is
           ``<patcher_name>?<arg1>&<arg2>=<val2>`` (positional arguments
           before keyword arguments).
        5) A list of the mixtures of the above.

    Returns:
      Value after applying the patchers. If any patcher returned a new value
      (by returning a single-item dict containing '' as its key), the return
      value will be a different object than ``value``; otherwise ``value``
      will be returned after applying the patches.

    Raises:
      ValueError: Error if the patch name and arguments cannot
        be parsed successfully.
    """
    patches = []
    rules = rule if isinstance(rule, list) else [rule]
    for p in rules:
        if isinstance(p, str):
            p = from_uri(p)
        if not isinstance(p, (Patcher, dict)) and not callable(p):
            raise TypeError(
                f'Patching rule {p!r} should be a dict of path to values, a rebind '
                f'function, a patcher (object or string), or a list of their '
                f'mixtures.'
            )
        patches.append(p)

    # Apply patches in chain, validating per OBJECT ERA (#773 ruling 9):
    # validators still run after the whole chain — so each catches later
    # patches clobbering its plan — but a step that REPLACES the value
    # outright starts a new era, dropping the validators of earlier
    # patchers rather than running them against an object they never saw.
    # A replacing patcher's own validator survives: it produced the new
    # object.
    pending_validators: list[Patcher] = []
    for p in patches:
        if isinstance(p, Patcher):
            new_value = p.patch(value)
            if new_value is not value:
                pending_validators.clear()
            value = new_value
            pending_validators.append(p)
        elif isinstance(p, dict):
            if len(p) == 1 and '' in p:
                value = p['']
                pending_validators.clear()
            else:
                value = value.sym_rebind(p)
        else:
            value = value.sym_rebind(p)

    for p in pending_validators:
        p.validate(value)
    return value

patcher

patcher(
    args: list[tuple[str, ValueSpec]] | None = None,
    name: str | None = None,
    auto_typing: bool = False,
    auto_doc: bool = False,
) -> Any

Decorate a function into a Patcher and register it.

A patcher function is defined as:

<patcher_fun> := <fun_name>(<target>, [parameters])

The signature takes at least one argument as the patching target, with additional arguments as patching parameters to control the details of this patch.

Example:

import pygx as pg

@pg.patching.patcher([
    ('x', pg.typing.Int())
])
def increment(v, x=1):
  return lambda k, v, p: v + x if isinstance(v, int) else v

d = pg.Dict(a=1)
pg.patching.patch(d, [increment(x=2)])   # d is now {a=3}.

# The same rule via a URI-like string. d is now {a=5}.
pg.patching.patch(d, ['increment?x=2'])

Parameters:

Name	Type	Description	Default
args	list[tuple[str, ValueSpec]] | None	A list of (arg_name, arg_value_spec) to schematize patcher arguments.	None
name	str | None	String to be used as the patcher name in URIs. If None, the function name will be used as the patcher name.	None
auto_typing	bool	If True, automatically infer the typing from the function signature.	False
auto_doc	bool	If True, use the docstring of the function as the docstring of the patcher.	False

Returns:

Type	Description
Any	A decorator that converts a function into a Patcher subclass.

Source code in pygx/patching/_rule_based.py

def patcher(
    args: list[tuple[str, pg_typing.ValueSpec]] | None = None,
    name: str | None = None,
    auto_typing: bool = False,
    auto_doc: bool = False,
) -> Any:
    """Decorate a function into a Patcher and register it.

    A patcher function is defined as:

    ```text
    <patcher_fun> := <fun_name>(<target>, [parameters])
    ```

    The signature takes at least one argument as the patching target,
    with additional arguments as patching parameters to control the details
    of this patch.

    **Example:**

    ```python
    import pygx as pg

    @pg.patching.patcher([
        ('x', pg.typing.Int())
    ])
    def increment(v, x=1):
      return lambda k, v, p: v + x if isinstance(v, int) else v

    d = pg.Dict(a=1)
    pg.patching.patch(d, [increment(x=2)])   # d is now {a=3}.

    # The same rule via a URI-like string. d is now {a=5}.
    pg.patching.patch(d, ['increment?x=2'])
    ```

    Args:
      args: A list of (arg_name, arg_value_spec) to schematize patcher
        arguments.
      name: String to be used as the patcher name in URIs. If None, the
        function name will be used as the patcher name.
      auto_typing: If True, automatically infer the typing from the function
        signature.
      auto_doc: If True, use the docstring of the function as the docstring
        of the patcher.

    Returns:
      A decorator that converts a function into a Patcher subclass.
    """
    functor_decorator = symbolic.functor(
        typing.cast(Any, args),
        base_class=Patcher,
        auto_typing=auto_typing,
        auto_doc=auto_doc,
    )

    def _decorator(fn):
        """Returns decorated Patcher class."""
        cls = functor_decorator(fn)
        _PATCHER_REGISTRY.register(
            name or fn.__name__, typing.cast(type[Patcher], cls)
        )
        arg_specs = cls.__signature__.args
        if len(arg_specs) < 1:
            raise TypeError(
                'Patcher function should have at least 1 argument '
                f'as patching target. (Patcher={cls.__type_name__!r})'
            )
        if not _is_patcher_target_spec(arg_specs[0].value_spec):
            raise TypeError(
                f'{arg_specs[0].value_spec!r} cannot be used for constraining '
                f'Patcher target. (Patcher={cls.__type_name__!r}, '
                f'Argument={arg_specs[0].name!r})\n'
                'Acceptable value spec types are: '
                'Any, Callable, Dict, Functor, List, Object.'
            )
        for arg_spec in arg_specs[1:]:
            if not _is_patcher_parameter_spec(arg_spec.value_spec):
                raise TypeError(
                    f'{arg_spec.value_spec!r} cannot be used for constraining '
                    f'Patcher argument. (Patcher={cls.__type_name__!r}, '
                    f'Argument={arg_spec.name!r})\n'
                    'Consider to treat it as string and parse yourself.'
                )
        return cls

    return _decorator

patcher_names

patcher_names()

Returns all registered patcher names.

Source code in pygx/patching/_rule_based.py

def patcher_names():
    """Returns all registered patcher names."""
    return _PATCHER_REGISTRY.names

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