`pygx.patching`¶

Systematic patching on symbolic values.

patching ¶

Systematic patching on symbolic values.

As pygx.Symbolic.rebind provides a flexible programming interface for modifying symbolic values, why bother to have this module? Here are the motivations:

Provide user friendly methods for addressing the most common patching patterns.
Provide a systematic solution for
- Patch semantic groups.
- Enable combination of these groups.
- Provide an interface that patching can be invoked 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 accomondate 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). Validation rule is a callable object that validate 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 pg.patcher decorator::

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

OR:

@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 instances of other patchers. return [ change_lr(lr), change_weight_decay(weight_decay) ]

After registration, a patcher object can be obtained from a URL-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 accomondate combination of semantic groups. A patcher in the sequence can propose updates to the original value or generate a replacement to 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 lookup 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:
      KeyError: constructor got unexpected 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 return the input itself unless fully replaced.

Source code in pygx/patching/_rule_based.py

def patch(self, x: symbolic.Symbolic) -> Any:
    """Patches an input and return the input itself unless fully replaced."""
    # 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 when a chain of patchers have been applied, as to validate the patched object in chain still conforms to the 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] when a chain of patchers
    have been applied, as to validate the patched object in chain still
    conforms to the 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 symbolic object from a base value and patches.

Parameters:

Name	Type	Description	Default
`value_type`	`type[Symbolic]`	Type of 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 `patch` for details.	`None`
`params_override`	`dict[str, Any] \| str \| None`	A rebind dict (or a JSON string as serialized rebind dict) as an additional patch to the value,	`None`

Returns:

Type	Description
`Any`	Value after applying `patchers` and `params_override` based on `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 symbolic object from a base value and patches.

    Args:
      value_type: Type of 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 `patch` for details.
      params_override: A rebind dict (or a JSON string as serialized rebind dict)
        as an additional patch to the value,

    Returns:
      Value after applying `patchers` 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)),
            raise_on_no_change=False,
        )
    return value

from_maybe_serialized ¶

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

Load value from maybe serialized form (e.g. JSON file or JSON string).

Parameters:

Name	Type	Description	Default
`source`	`Any \| str`	Source of value. It can be value (non-string type) itself, or a filepath, or a JSON string from where 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.

Source code in pygx/patching/_object_factory.py

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

    Args:
      source: Source of value. It can be value (non-string type) itself, or a
        filepath, or a JSON string from where the value will be loaded.
      value_type: An optional type to constrain the value.

    Returns:
      Value from source.
    """
    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 = None,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values on matched keys (leaf-node names).

Example::

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

print(pg.patching.patch_on_key(d, '.', value=3)) # {a=3, b=3}

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() 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_keys(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 key name.	required
`value`	`Any`	New value for field that satisfy `condition`.	`None`
`value_fn`	`Callable[[Any], Any] \| None`	Callable object that produces new value based on old value. If not None, `value` must be None.	`None`
`skip_notification`	`bool \| None`	If True, `on_change` event will not be triggered for this operation. If None, the behavior is decided by `pg.notify_on_rebind`. Please see `symbolic.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 = None,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values on matched keys (leaf-node names).

    Example::

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

      print(pg.patching.patch_on_key(d, '.', value=3))
      # {a=3, b=3}

      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()
      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_keys(a, 'x', value=4, skip_notification=True)
      # a._num_changes is still 2.

    Args:
      src: symbolic value to patch.
      regex: Regex for key name.
      value: New value for field that satisfy `condition`.
      value_fn: Callable object that produces new value based on old value.
        If not None, `value` must be None.
      skip_notification: If True, `on_change` event will not be triggered for this
        operation. If None, the behavior is decided by `pg.notify_on_rebind`.
        Please see `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 = None,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any

Recursively patch values that are the requested member of classes.

Example::

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

Parameters:

Name	Type	Description	Default
`src`	`Symbolic`	symbolic value to patch.	required
`cls`	`type[Any] \| tuple[type[Any], ...]`	In which class the member belongs to.	required
`name`	`str`	Member name.	required
`value`	`Any`	New value for field that satisfy `condition`.	`None`
`value_fn`	`Callable[[Any], Any] \| None`	Callable object that produces new value based on old value. If not None, `value` must be None.	`None`
`skip_notification`	`bool \| None`	If True, `on_change` event will not be triggered for this operation. If None, the behavior is decided by `pg.notify_on_rebind`. Please see `symbolic.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 = None,
    value_fn: Callable[[Any], Any] | None = None,
    skip_notification: bool | None = None,
) -> Any:
    """Recursively patch values that are the requested member of classes.

    Example::

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

    Args:
      src: symbolic value to patch.
      cls: In which class the member belongs to.
      name: Member name.
      value: New value for field that satisfy `condition`.
      value_fn: Callable object that produces new value based on old value.
        If not None, `value` must be None.
      skip_notification: If True, `on_change` event will not be triggered for this
        operation. If None, the behavior is decided by `pg.notify_on_rebind`.
        Please see `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 = None,
    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) print(pg.patching.patch_on_path(d, '.*x', value=3)) # {a={x=1}, b=2}

Parameters:

Name	Type	Description	Default
`src`	`Symbolic`	symbolic value to patch.	required
`regex`	`str`	Regex for key path.	required
`value`	`Any`	New value for field that satisfy `condition`.	`None`
`value_fn`	`Callable[[Any], Any] \| None`	Callable object that produces new value based on old value. If not None, `value` must be None.	`None`
`skip_notification`	`bool \| None`	If True, `on_change` event will not be triggered for this operation. If None, the behavior is decided by `pg.notify_on_rebind`. Please see `symbolic.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 = None,
    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)
      print(pg.patching.patch_on_path(d, '.*x', value=3))
      # {a={x=1}, b=2}

    Args:
      src: symbolic value to patch.
      regex: Regex for key path.
      value: New value for field that satisfy `condition`.
      value_fn: Callable object that produces new value based on old value.
        If not None, `value` must be None.
      skip_notification: If True, `on_change` event will not be triggered for this
        operation. If None, the behavior is decided by `pg.notify_on_rebind`.
        Please see `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 = None,
    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) print(pg.patching.patch_on_type(d, int, value_fn=lambda x: x * 2)) # {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 field that satisfy `condition`.	`None`
`value_fn`	`Callable[[Any], Any] \| None`	Callable object that produces new value based on old value. If not None, `value` must be None.	`None`
`skip_notification`	`bool \| None`	If True, `on_change` event will not be triggered for this operation. If None, the behavior is decided by `pg.notify_on_rebind`. Please see `symbolic.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 = None,
    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)
      print(pg.patching.patch_on_type(d, int, value_fn=lambda x: x * 2))
      # {a={x=2}, b=4}

    Args:
      src: symbolic value to patch.
      value_type: Value type to match.
      value: New value for field that satisfy `condition`.
      value_fn: Callable object that produces new value based on old value.
        If not None, `value` must be None.
      skip_notification: If True, `on_change` event will not be triggered for this
        operation. If None, the behavior is decided by `pg.notify_on_rebind`.
        Please see `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 = None,
    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) print(pg.patching.patch_on_value(d, 1, value=3)) # {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 field that satisfy `condition`.	`None`
`value_fn`	`Callable[[Any], Any] \| None`	Callable object that produces new value based on old value. If not None, `value` must be None.	`None`
`skip_notification`	`bool \| None`	If True, `on_change` event will not be triggered for this operation. If None, the behavior is decided by `pg.notify_on_rebind`. Please see `symbolic.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 = None,
    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)
      print(pg.patching.patch_on_value(d, 1, value=3))
      # {a={x=3}, b=3}

    Args:
      src: symbolic value to patch.
      old_value: Old value to match.
      value: New value for field that satisfy `condition`.
      value_fn: Callable object that produces new value based on old value.
        If not None, `value` must be None.
      skip_notification: If True, `on_change` event will not be triggered for this
        operation. If None, the behavior is decided by `pg.notify_on_rebind`.
        Please see `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)

If True, allow registration with the same patch name.

Source code in pygx/patching/_rule_based.py

def allow_repeated_patcher_registration(allow: bool = True):
    """If True, allow registration with the same patch name."""
    global _ALLOW_REPEATED_PATCHER_REGISTRATION  # pylint: disable=global-statement
    _ALLOW_REPEATED_PATCHER_REGISTRATION = allow

from_uri ¶

from_uri(uri: str) -> Patcher

Create a Patcher object from a URI-like string.

Source code in pygx/patching/_rule_based.py

def from_uri(uri: str) -> Patcher:
    """Create a Patcher object from a URI-like string."""
    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.rebind`. 3) A `pygx.patching.Patcher` object. 4) A URL-like string representing an instance of a register Patcher. The format is "?&=". 5) A list of the mixtures of 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 that containing '' as key), the return
`Any`	value will be a different object other 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.rebind`][pygx.symbolic.Symbolic.sym_rebind].
        3) A [`pygx.patching.Patcher`][pygx.patching.Patcher] object.
        4) A URL-like string representing an instance of a register Patcher.
           The format is "<patcher_name>?<arg1>&<arg2>=<val2>".
        5) A list of the mixtures of above.

    Returns:
      Value after applying the patchers. If any patcher returned a new value
      (by returning a single-item dict that containing '' as key), the return
      value will be a different object other 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.
    for p in patches:
        if isinstance(p, Patcher):
            value = p.patch(value)
        elif isinstance(p, dict):
            if len(p) == 1 and '' in p:
                value = p['']
            else:
                value = value.sym_rebind(p, raise_on_no_change=False)
        else:
            value = value.sym_rebind(p, raise_on_no_change=False)

    # Validate patched values.
    for p in patches:
        if isinstance(p, Patcher):
            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::

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

# This patcher can be called via: # pg.patching.apply(v, [increment(x=2)]) # or pg.patching.apply(v, ['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 patcher name in URI. If None, function name will be used as patcher name.	`None`
`auto_typing`	`bool`	If True, automatically inference 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:

       `<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::

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

      # This patcher can be called via:
      # pg.patching.apply(v, [increment(x=2)])
      # or pg.patching.apply(v, ['increment?x=2'])

    Args:
      args: A list of (arg_name, arg_value_spec) to schematize patcher arguments.
      name: String to be used as patcher name in URI. If None, function name will
        be used as patcher name.
      auto_typing: If True, automatically inference 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 patch names.

Source code in pygx/patching/_rule_based.py

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

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

pygx.patching¶

patching ¶

Patcher ¶

patch ¶

validate ¶

ObjectFactory ¶

from_maybe_serialized ¶

patch_on_key ¶

patch_on_member ¶

patch_on_path ¶

patch_on_type ¶

patch_on_value ¶

allow_repeated_patcher_registration ¶

from_uri ¶

patch ¶

patcher ¶

patcher_names ¶

`pygx.patching`¶