Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 16 additions & 9 deletions src/psygnal/_evented_decorator.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,16 @@
from __future__ import annotations

from typing import (
Any,
Callable,
Literal,
TypeVar,
overload,
)
from typing import TYPE_CHECKING, Callable, Literal, Mapping, TypeVar, overload

from psygnal._group_descriptor import SignalGroupDescriptor

if TYPE_CHECKING:
from psygnal._group_descriptor import EqOperator, FieldAliasFunc

__all__ = ["evented"]

T = TypeVar("T", bound=type)

EqOperator = Callable[[Any, Any], bool]


@overload
def evented(
Expand All @@ -25,6 +20,7 @@ def evented(
equality_operators: dict[str, EqOperator] | None = None,
warn_on_no_fields: bool = ...,
cache_on_instance: bool = ...,
signal_aliases: Mapping[str, str | None] | FieldAliasFunc | None = ...,
) -> T: ...


Expand All @@ -36,6 +32,7 @@ def evented(
equality_operators: dict[str, EqOperator] | None = None,
warn_on_no_fields: bool = ...,
cache_on_instance: bool = ...,
signal_aliases: Mapping[str, str | None] | FieldAliasFunc | None = ...,
) -> Callable[[T], T]: ...


Expand All @@ -46,6 +43,7 @@ def evented(
equality_operators: dict[str, EqOperator] | None = None,
warn_on_no_fields: bool = True,
cache_on_instance: bool = True,
signal_aliases: Mapping[str, str | None] | FieldAliasFunc | None = None,
) -> Callable[[T], T] | T:
"""A decorator to add events to a dataclass.

Expand Down Expand Up @@ -85,6 +83,14 @@ def evented(
access, but means that the owner instance will no longer be pickleable. If
`False`, the SignalGroup instance will *still* be cached, but not on the
instance itself.
signal_aliases: Mapping[str, str | None] | Callable[[str], str | None] | None
If defined, a mapping between field name and signal name. Field names that are
not `signal_aliases` keys are not aliased (the signal name is the field name).
If the dict value is None, do not create a signal associated with this field.
If a callable, the signal name is the output of the function applied to the
field name. If the output is None, no signal is created for this field.
If None, defaults to an empty dict, no aliases.
Default to None

Returns
-------
Expand Down Expand Up @@ -122,6 +128,7 @@ def _decorate(cls: T) -> T:
equality_operators=equality_operators,
warn_on_no_fields=warn_on_no_fields,
cache_on_instance=cache_on_instance,
signal_aliases=signal_aliases,
)
# as a decorator, this will have already been called
descriptor.__set_name__(cls, events_namespace)
Expand Down
16 changes: 15 additions & 1 deletion src/psygnal/_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,7 @@ class MySignals(SignalGroup):
_psygnal_signals: ClassVar[Mapping[str, Signal]]
_psygnal_uniform: ClassVar[bool] = False
_psygnal_name_conflicts: ClassVar[set[str]]
_psygnal_aliases: ClassVar[dict[str, str | None]]

_psygnal_instances: dict[str, SignalInstance]

Expand All @@ -280,7 +281,11 @@ def __init__(self, instance: Any = None) -> None:
}
self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)

def __init_subclass__(cls, strict: bool = False) -> None:
def __init_subclass__(
cls,
strict: bool = False,
signal_aliases: Mapping[str, str | None] = {},
) -> None:
"""Collects all Signal instances on the class under `cls._psygnal_signals`."""
# Collect Signals and remove from class attributes
# Use dir(cls) instead of cls.__dict__ to get attributes from super()
Expand Down Expand Up @@ -328,6 +333,9 @@ def __init_subclass__(cls, strict: bool = False) -> None:
stacklevel=2,
)

aliases = getattr(cls, "_psygnal_aliases", {})
cls._psygnal_aliases = {**aliases, **signal_aliases}

cls._psygnal_uniform = _is_uniform(cls._psygnal_signals.values())
if strict and not cls._psygnal_uniform:
raise TypeError(
Expand Down Expand Up @@ -403,6 +411,12 @@ def __repr__(self) -> str:
name = self.__class__.__name__
return f"<SignalGroup {name!r} with {len(self)} signals>"

def get_signal_by_alias(self, name: str) -> SignalInstance | None:
sig_name = self._psygnal_aliases.get(name, name)
if sig_name is None or sig_name not in self:
return None
return self[sig_name]

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hey @getzze, I think this is probably my only question (in an extremely nice PR). I can see why you did it this way as opposed to modifing __getitem__ It allows someone to query both the alias name and the original name. Do you think that's important? (I assume so, since you've been giving it a lot of thought and opted for this).

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hey, I think it can be removed, I mean inlined in evented_setattr, as it is only used there. And SignalGroup has one less method.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i do think that end-users should also be able to search by alias. but i guess the question is should they no longer be able to search by the original name? i.e. if I remove the method on SignalGroup, then end-users can no longer access the original names for an aliased signal, correct? are we ok with that?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I made a function instead of a method, with a new name get_signal_from_field.


@classmethod
def psygnals_uniform(cls) -> bool:
"""Return true if all signals in the group have the same signature."""
Expand Down
132 changes: 113 additions & 19 deletions src/psygnal/_group_descriptor.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
ClassVar,
Iterable,
Literal,
Mapping,
Type,
TypeVar,
cast,
Expand All @@ -28,14 +29,16 @@

from psygnal._weak_callback import RefErrorChoice, WeakCallback

EqOperator = Callable[[Any, Any], bool]
FieldAliasFunc = Callable[[str], str | None]

__all__ = ["is_evented", "get_evented_namespace", "SignalGroupDescriptor"]


T = TypeVar("T", bound=Type)
S = TypeVar("S")


EqOperator = Callable[[Any, Any], bool]
_EQ_OPERATORS: dict[type, dict[str, EqOperator]] = {}
_EQ_OPERATOR_NAME = "__eq_operators__"
PSYGNAL_GROUP_NAME = "_psygnal_group_"
Expand Down Expand Up @@ -146,6 +149,7 @@ def _build_dataclass_signal_group(
cls: type,
signal_group_class: type[SignalGroup],
equality_operators: Iterable[tuple[str, EqOperator]] | None = None,
signal_aliases: Mapping[str, str | None] | FieldAliasFunc | None = None,
) -> type[SignalGroup]:
"""Build a SignalGroup with events for each field in a dataclass.

Expand All @@ -160,10 +164,35 @@ def _build_dataclass_signal_group(
If defined, a mapping of field name and equality operator to use to compare if
each field was modified after being set.
Default to None
signal_aliases: Mapping[str, str | None] | Callable[[str], str | None] | None
If defined, a mapping between field name and signal name. Field names that are
not `signal_aliases` keys are not aliased (the signal name is the field name).
If the dict value is None, do not create a signal associated with this field.
If a callable, the signal name is the output of the function applied to the
field name. If the output is None, no signal is created for this field.
If None, defaults to an empty dict, no aliases.
Default to None

"""
group_name = f"{cls.__name__}{signal_group_class.__name__}"
# parse arguments
_equality_operators = dict(equality_operators) if equality_operators else {}
signals = {}
eq_map = _get_eq_operator_map(cls)

# prepare signal_aliases lookup
transform: FieldAliasFunc | None = None
_signal_aliases: dict[str, str | None] = {}
if callable(signal_aliases):
transform = signal_aliases
else:
_signal_aliases = dict(signal_aliases) if signal_aliases else {}
signal_group_sig_names = list(getattr(signal_group_class, "_psygnal_signals", {}))
signal_group_sig_aliases = cast(
"dict[str, str | None]",
dict(getattr(signal_group_class, "_psygnal_aliases", {})),
)

signals = {}
# create a Signal for each field in the dataclass
for name, type_ in iter_fields(cls):
if name in _equality_operators:
Expand All @@ -172,14 +201,54 @@ def _build_dataclass_signal_group(
eq_map[name] = _equality_operators[name]
else:
eq_map[name] = _pick_equality_operator(type_)

# Resolve the signal name for the field
sig_name: str | None
if name in _signal_aliases: # an alias has been provided in a mapping
sig_name = _signal_aliases[name]
elif callable(transform): # a callable has been provided
sig_name = transform(name)
elif name in signal_group_sig_aliases: # an alias has been defined in the class
sig_name = signal_group_sig_aliases[name]
else: # no alias has been defined, use the field name as the signal name
sig_name = name

# Add the field and signal name to the table of signals, to emit with `setattr`
_signal_aliases[name] = sig_name

# An alias mapping or callable returned `None`, skip this field
if sig_name is None:
continue

# Repeated signal
if sig_name in signals:
key = next((k for k, v in _signal_aliases.items() if v == sig_name), None)
warnings.warn(
f"Skip signal {sig_name!r}, was already created in {group_name}, "
f"from field {key}",
UserWarning,
stacklevel=2,
)
continue
if sig_name in signal_group_sig_names:
warnings.warn(
f"Skip signal {sig_name!r}, was already defined by "
f"{signal_group_class}",
UserWarning,
stacklevel=2,
)
continue

# Create the Signal
field_type = object if type_ is None else type_
signals[name] = sig = Signal(field_type, field_type)
signals[sig_name] = sig = Signal(field_type, field_type)
# patch in our custom SignalInstance class with maxargs=1 on connect_setattr
sig._signal_instance_class = _DataclassFieldSignalInstance

# Create `signal_group_class` subclass with the attached signals
group_name = f"{cls.__name__}{signal_group_class.__name__}"
return type(group_name, (signal_group_class,), signals)
# Create `signal_group_class` subclass with the attached signals and signal_aliases
return type(
group_name, (signal_group_class,), signals, signal_aliases=_signal_aliases
)


def is_evented(obj: object) -> bool:
Expand Down Expand Up @@ -286,12 +355,12 @@ def _setattr_and_emit_(self: object, name: str, value: Any) -> None:
return super_setattr(self, name, value)

group: SignalGroup | None = getattr(self, signal_group_name, None)
if not isinstance(group, SignalGroup) or name not in group:
return super_setattr(self, name, value)
if not isinstance(group, SignalGroup):
return super_setattr(self, name, value) # pragma: no cover

# don't emit if the signal doesn't exist or has no listeners
signal: SignalInstance = group[name]
if len(signal) < 1:
signal: SignalInstance | None = group.get_signal_by_alias(name)
if signal is None or len(signal) < 1:
return super_setattr(self, name, value)

with _changes_emitted(self, name, signal):
Expand Down Expand Up @@ -374,6 +443,14 @@ def __setattr__(self, name: str, value: Any) -> None:
instance will be a subclass of `signal_group_class` (SignalGroup if it is None).
If False, a deepcopy of `signal_group_class` will be used.
Default to True
signal_aliases: Mapping[str, str | None] | Callable[[str], str | None] | None
If defined, a mapping between field name and signal name. Field names that are
not `signal_aliases` keys are not aliased (the signal name is the field name).
If the dict value is None, do not create a signal associated with this field.
If a callable, the signal name is the output of the function applied to the
field name. If the output is None, no signal is created for this field.
If None, defaults to an empty dict, no aliases.
Default to None

Examples
--------
Expand Down Expand Up @@ -409,27 +486,36 @@ def __init__(
patch_setattr: bool = True,
signal_group_class: type[SignalGroup] | None = None,
collect_fields: bool = True,
signal_aliases: Mapping[str, str | None] | FieldAliasFunc | None = None,
):
grp_cls = signal_group_class or SignalGroup
if not (isinstance(grp_cls, type) and issubclass(grp_cls, SignalGroup)):
raise TypeError( # pragma: no cover
f"'signal_group_class' must be a subclass of SignalGroup, "
f"not {grp_cls}"
)
if grp_cls is SignalGroup and collect_fields is False:
raise ValueError(
"Cannot use SignalGroup with collect_fields=False. "
"Use a custom SignalGroup subclass instead."
)
if not collect_fields:
if grp_cls is SignalGroup:
raise ValueError(
"Cannot use SignalGroup with `collect_fields=False`. "
"Use a custom SignalGroup subclass instead."
)

if callable(signal_aliases):
raise ValueError(
"Cannot use a Callable for `signal_aliases` with "
"`collect_fields=False`"
)

self._name: str | None = None
self._eqop = tuple(equality_operators.items()) if equality_operators else None
self._warn_on_no_fields = warn_on_no_fields
self._cache_on_instance = cache_on_instance
self._patch_setattr = patch_setattr

self._signal_group_class: type[SignalGroup] = grp_cls
self._collect_fields = collect_fields
self._signal_aliases = signal_aliases

self._signal_groups: dict[int, type[SignalGroup]] = {}

def __set_name__(self, owner: type, name: str) -> None:
Expand Down Expand Up @@ -500,15 +586,23 @@ def _get_signal_group(self, owner: type) -> type[SignalGroup]:
return self._signal_groups[type_id]

def _create_group(self, owner: type) -> type[SignalGroup]:
# Do not collect fields from owner class, copy the SignalGroup
if not self._collect_fields:
# Do not collect fields from owner class
Group = copy.deepcopy(self._signal_group_class)

# Collect fields and create SignalGroup subclass
# Add aliases
if isinstance(self._signal_aliases, dict):
Group._psygnal_aliases.update(self._signal_aliases)

else:
# Collect fields and create SignalGroup subclass
Group = _build_dataclass_signal_group(
owner, self._signal_group_class, equality_operators=self._eqop
owner,
self._signal_group_class,
equality_operators=self._eqop,
signal_aliases=self._signal_aliases,
)

if self._warn_on_no_fields and not Group._psygnal_signals:
warnings.warn(
f"No mutable fields found on class {owner}: no events will be "
Expand Down
5 changes: 2 additions & 3 deletions tests/test_evented_model.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,9 +70,8 @@ class User(EventedModel):

# test event system
assert isinstance(user.events, SignalGroup)
# with pytest.warns(FutureWarning):
assert "id" in user.events.signals
assert "name" in user.events.signals
assert "id" in user.events
assert "name" in user.events

# ClassVars are excluded from events
assert "age" not in user.events
Expand Down
Loading