Skip to content

Core API#

psygnal.Signal #

Declares a signal emitter on a class.

This is class implements the descriptor protocol and is designed to be used as a class attribute, with the supported signature types provided in the constructor:

from psygnal import Signal

class MyEmitter:
    changed = Signal(int)

def receiver(arg: int):
    print("new value:", arg)

emitter = MyEmitter()
emitter.changed.connect(receiver)
emitter.changed.emit(1)  # prints 'new value: 1'

Note

in the example above, MyEmitter.changed is an instance of Signal, and emitter.changed is an instance of SignalInstance. See the documentation on SignalInstance for details on how to connect to and/or emit a signal on an instance of an object that has a Signal.

Parameters:

  • *types (Union[Type[Any], Signature]) –

    A sequence of individual types, or a single inspect.Signature object.

  • description (str) –

    Optional descriptive text for the signal. (not used internally).

  • name (Optional[str]) –

    Optional name of the signal. If it is not specified then the name of the class attribute that is bound to the signal will be used. default None

  • check_nargs_on_connect (bool) –

    Whether to check the number of positional args against signature when connecting a new callback. This can also be provided at connection time using .connect(..., check_nargs=True). By default, True.

  • check_types_on_connect (bool) –

    Whether to check the callback parameter types against signature when connecting a new callback. This can also be provided at connection time using .connect(..., check_types=True). By default, False.

Source code in psygnal/_signal.py
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
class Signal:
    """Declares a signal emitter on a class.

    This is class implements the [descriptor
    protocol](https://docs.python.org/3/howto/descriptor.html#descriptorhowto)
    and is designed to be used as a class attribute, with the supported signature types
    provided in the constructor:

    ```python
    from psygnal import Signal

    class MyEmitter:
        changed = Signal(int)

    def receiver(arg: int):
        print("new value:", arg)

    emitter = MyEmitter()
    emitter.changed.connect(receiver)
    emitter.changed.emit(1)  # prints 'new value: 1'
    ```

    !!! note

        in the example above, `MyEmitter.changed` is an instance of `Signal`,
        and `emitter.changed` is an instance of `SignalInstance`.  See the
        documentation on [`SignalInstance`][psygnal.SignalInstance] for details
        on how to connect to and/or emit a signal on an instance of an object
        that has a `Signal`.


    Parameters
    ----------
    *types : Union[Type[Any], Signature]
        A sequence of individual types, or a *single* [`inspect.Signature`][] object.
    description : str
        Optional descriptive text for the signal.  (not used internally).
    name : Optional[str]
        Optional name of the signal. If it is not specified then the name of the
        class attribute that is bound to the signal will be used. default None
    check_nargs_on_connect : bool
        Whether to check the number of positional args against `signature` when
        connecting a new callback. This can also be provided at connection time using
        `.connect(..., check_nargs=True)`. By default, True.
    check_types_on_connect : bool
        Whether to check the callback parameter types against `signature` when
        connecting a new callback. This can also be provided at connection time using
        `.connect(..., check_types=True)`. By default, False.
    """

    # _signature: Signature  # callback signature for this signal

    _current_emitter: ClassVar[SignalInstance | None] = None

    def __init__(
        self,
        *types: type[Any] | Signature,
        description: str = "",
        name: str | None = None,
        check_nargs_on_connect: bool = True,
        check_types_on_connect: bool = False,
    ) -> None:
        self._name = name
        self.description = description
        self._check_nargs_on_connect = check_nargs_on_connect
        self._check_types_on_connect = check_types_on_connect
        self._signal_instance_class: type[SignalInstance] = SignalInstance

        if types and isinstance(types[0], Signature):
            self._signature = types[0]
            if len(types) > 1:
                warnings.warn(
                    "Only a single argument is accepted when directly providing a"
                    f" `Signature`.  These args were ignored: {types[1:]}",
                    stacklevel=2,
                )
        else:
            self._signature = _build_signature(*cast("tuple[Type[Any], ...]", types))

    @property
    def signature(self) -> Signature:
        """[Signature][inspect.Signature] supported by this Signal."""
        return self._signature

    def __set_name__(self, owner: type[Any], name: str) -> None:
        """Set name of signal when declared as a class attribute on `owner`."""
        if self._name is None:
            self._name = name

    @overload
    def __get__(
        self, instance: None, owner: type[Any] | None = None
    ) -> Signal: ...  # pragma: no cover

    @overload
    def __get__(
        self, instance: Any, owner: type[Any] | None = None
    ) -> SignalInstance: ...  # pragma: no cover

    def __get__(
        self, instance: Any, owner: type[Any] | None = None
    ) -> Signal | SignalInstance:
        """Get signal instance.

        This is called when accessing a Signal instance.  If accessed as an
        attribute on the class `owner`, instance, will be `None`.  Otherwise,
        if `instance` is not None, we're being accessed on an instance of `owner`.

            class Emitter:
                signal = Signal()

            e = Emitter()

            E.signal  # instance will be None, owner will be Emitter
            e.signal  # instance will be e, owner will be Emitter

        Returns
        -------
        Signal or SignalInstance
            Depending on how this attribute is accessed.
        """
        if instance is None:
            return self
        name = cast("str", self._name)
        signal_instance = self._signal_instance_class(
            self.signature,
            instance=instance,
            name=name,
            check_nargs_on_connect=self._check_nargs_on_connect,
            check_types_on_connect=self._check_types_on_connect,
        )
        # instead of caching this signal instance on self, we just assign it
        # to instance.name ... this essentially breaks the descriptor,
        # (i.e. __get__ will never again be called for this instance, and we have no
        # idea how many instances are out there),
        # but it allows us to prevent creating a key for this instance (which may
        # not be hashable or weak-referenceable), and also provides a significant
        # speedup on attribute access (affecting everything).
        # (note, this is the same mechanism used in the `cached_property` decorator)
        try:
            setattr(instance, name, signal_instance)
        except AttributeError as e:
            from ._group import SignalGroup

            if name == "all" and isinstance(instance, SignalGroup):
                # this specific case will happen if an evented dataclass field is named
                # "all". 'all' is a reserved name for the SignalRelay, but we've
                # already caught and warned about it in SignalGroup.__init_subclass__.
                pass
            else:
                # otherwise, give an informative error message
                raise AttributeError(  # pragma: no cover
                    "An attempt to cache a SignalInstance on instance "
                    f"{instance} failed. Please report this with your use case at "
                    "https://github.com/pyapp-kit/psygnal/issues."
                ) from e

        return signal_instance

    @classmethod
    @contextmanager
    def _emitting(cls, emitter: SignalInstance) -> Iterator[None]:
        """Context that sets the sender on a receiver object while emitting a signal."""
        previous, cls._current_emitter = cls._current_emitter, emitter
        try:
            yield
        finally:
            cls._current_emitter = previous

    @classmethod
    def current_emitter(cls) -> SignalInstance | None:
        """Return currently emitting `SignalInstance`, if any.

        This will typically be used in a callback.

        Examples
        --------
        ```python
        from psygnal import Signal

        def my_callback():
            source = Signal.current_emitter()
        ```
        """
        return cls._current_emitter

    @classmethod
    def sender(cls) -> Any:
        """Return currently emitting object, if any.

        This will typically be used in a callback.
        """
        return getattr(cls._current_emitter, "instance", None)

signature: Signature property #

Signature supported by this Signal.

current_emitter() classmethod #

Return currently emitting SignalInstance, if any.

This will typically be used in a callback.

Examples:

from psygnal import Signal

def my_callback():
    source = Signal.current_emitter()
Source code in psygnal/_signal.py
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@classmethod
def current_emitter(cls) -> SignalInstance | None:
    """Return currently emitting `SignalInstance`, if any.

    This will typically be used in a callback.

    Examples
    --------
    ```python
    from psygnal import Signal

    def my_callback():
        source = Signal.current_emitter()
    ```
    """
    return cls._current_emitter

sender() classmethod #

Return currently emitting object, if any.

This will typically be used in a callback.

Source code in psygnal/_signal.py
243
244
245
246
247
248
249
@classmethod
def sender(cls) -> Any:
    """Return currently emitting object, if any.

    This will typically be used in a callback.
    """
    return getattr(cls._current_emitter, "instance", None)

psygnal.SignalInstance #

A signal instance (optionally) bound to an object.

In most cases, users will not create a SignalInstance directly -- instead creating a Signal class attribute. This object will be instantiated by the Signal.__get__ method (i.e. the descriptor protocol), when a Signal instance is accessed from an instance of a class with Signal attribute.

However, it is the SignalInstance that you will most often be interacting with when you access the name of a Signal on an instance -- so understanding the SignalInstance API is key to using psygnal.

class Emitter:
    signal = Signal()

e = Emitter()

# when accessed on an *instance* of Emitter,
# the signal attribute will be a SignalInstance
e.signal

# This is what you will use to connect your callbacks
e.signal.connect(some_callback)

Parameters:

  • signature (Optional[inspect.Signature]) –

    The signature that this signal accepts and will emit, by default Signature().

  • instance (Optional[Any]) –

    An object to which this signal is bound. Normally this will be provided by the Signal.__get__ method (see above). However, an unbound SignalInstance may also be created directly. by default None.

  • name (Optional[str]) –

    An optional name for this signal. Normally this will be provided by the Signal.__get__ method. by default None

  • check_nargs_on_connect (bool) –

    Whether to check the number of positional args against signature when connecting a new callback. This can also be provided at connection time using .connect(..., check_nargs=True). By default, True.

  • check_types_on_connect (bool) –

    Whether to check the callback parameter types against signature when connecting a new callback. This can also be provided at connection time using .connect(..., check_types=True). By default, False.

Raises:

  • TypeError

    If signature is neither an instance of inspect.Signature, or a tuple of types.

Source code in psygnal/_signal.py
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
@mypyc_attr(allow_interpreted_subclasses=True)
class SignalInstance:
    """A signal instance (optionally) bound to an object.

    In most cases, users will not create a `SignalInstance` directly -- instead
    creating a [Signal][psygnal.Signal] class attribute.  This object will be
    instantiated by the `Signal.__get__` method (i.e. the descriptor protocol),
    when a `Signal` instance is accessed from an *instance* of a class with `Signal`
    attribute.

    However, it is the `SignalInstance` that you will most often be interacting
    with when you access the name of a `Signal` on an instance -- so understanding
    the `SignalInstance` API is key to using psygnal.

    ```python
    class Emitter:
        signal = Signal()

    e = Emitter()

    # when accessed on an *instance* of Emitter,
    # the signal attribute will be a SignalInstance
    e.signal

    # This is what you will use to connect your callbacks
    e.signal.connect(some_callback)
    ```

    Parameters
    ----------
    signature : Optional[inspect.Signature]
        The signature that this signal accepts and will emit, by default `Signature()`.
    instance : Optional[Any]
        An object to which this signal is bound. Normally this will be provided by the
        `Signal.__get__` method (see above).  However, an unbound `SignalInstance`
        may also be created directly. by default `None`.
    name : Optional[str]
        An optional name for this signal.  Normally this will be provided by the
        `Signal.__get__` method. by default `None`
    check_nargs_on_connect : bool
        Whether to check the number of positional args against `signature` when
        connecting a new callback. This can also be provided at connection time using
        `.connect(..., check_nargs=True)`. By default, True.
    check_types_on_connect : bool
        Whether to check the callback parameter types against `signature` when
        connecting a new callback. This can also be provided at connection time using
        `.connect(..., check_types=True)`. By default, False.

    Raises
    ------
    TypeError
        If `signature` is neither an instance of `inspect.Signature`, or a `tuple`
        of types.
    """

    _is_blocked: bool = False
    _is_paused: bool = False
    _debug_hook: ClassVar[Callable[[EmissionInfo], None] | None] = None

    def __init__(
        self,
        signature: Signature | tuple = _empty_signature,
        *,
        instance: Any = None,
        name: str | None = None,
        check_nargs_on_connect: bool = True,
        check_types_on_connect: bool = False,
    ) -> None:
        self._name = name
        self._instance: Callable = self._instance_ref(instance)
        self._args_queue: list[tuple] = []  # filled when paused

        if isinstance(signature, (list, tuple)):
            signature = _build_signature(*signature)
        elif not isinstance(signature, Signature):  # pragma: no cover
            raise TypeError(
                "`signature` must be either a sequence of types, or an "
                "instance of `inspect.Signature`"
            )

        self._signature = signature
        self._check_nargs_on_connect = check_nargs_on_connect
        self._check_types_on_connect = check_types_on_connect
        self._slots: list[WeakCallback] = []
        self._is_blocked: bool = False
        self._is_paused: bool = False
        self._lock = threading.RLock()

    @staticmethod
    def _instance_ref(instance: Any) -> Callable[[], Any]:
        if instance is None:
            return lambda: None

        try:
            return weakref.ref(instance)
        except TypeError:
            # fall back to strong reference if instance is not weak-referenceable
            return lambda: instance

    @property
    def signature(self) -> Signature:
        """Signature supported by this `SignalInstance`."""
        return self._signature

    @property
    def instance(self) -> Any:
        """Object that emits this `SignalInstance`."""
        return self._instance()

    @property
    def name(self) -> str:
        """Name of this `SignalInstance`."""
        return self._name or ""

    def __repr__(self) -> str:
        """Return repr."""
        name = f" {self._name!r}" if self._name else ""
        instance = f" on {self.instance!r}" if self.instance is not None else ""
        return f"<{type(self).__name__}{name}{instance}>"

    @overload
    def connect(
        self,
        *,
        thread: threading.Thread | Literal["main", "current"] | None = ...,
        check_nargs: bool | None = ...,
        check_types: bool | None = ...,
        unique: bool | str = ...,
        max_args: int | None = None,
        on_ref_error: RefErrorChoice = ...,
    ) -> Callable[[F], F]: ...  # pragma: no cover

    @overload
    def connect(
        self,
        slot: F,
        *,
        thread: threading.Thread | Literal["main", "current"] | None = ...,
        check_nargs: bool | None = ...,
        check_types: bool | None = ...,
        unique: bool | str = ...,
        max_args: int | None = None,
        on_ref_error: RefErrorChoice = ...,
    ) -> F: ...  # pragma: no cover

    def connect(
        self,
        slot: F | None = None,
        *,
        thread: threading.Thread | Literal["main", "current"] | None = None,
        check_nargs: bool | None = None,
        check_types: bool | None = None,
        unique: bool | str = False,
        max_args: int | None = None,
        on_ref_error: RefErrorChoice = "warn",
    ) -> Callable[[F], F] | F:
        """Connect a callback (`slot`) to this signal.

        `slot` is compatible if:

        * it requires no more than the number of positional arguments emitted by this
          `SignalInstance`.  (It *may* require less)
        * it has no *required* keyword arguments (keyword only arguments that have
          no default).
        * if `check_types` is `True`, the parameter types in the callback signature must
          match the signature of this `SignalInstance`.

        This method may be used as a decorator.

        ```python
        @signal.connect
        def my_function():
            ...
        ```

        !!!important
            If a signal is connected with `thread != None`, then it is up to the user
            to ensure that `psygnal.emit_queued` is called, or that one of the backend
            convenience functions is used (e.g. `psygnal.qt.start_emitting_from_queue`).
            Otherwise, callbacks that are connected to signals that are emitted from
            another thread will never be called.

        Parameters
        ----------
        slot : Callable
            A callable to connect to this signal.  If the callable accepts less
            arguments than the signature of this slot, then they will be discarded when
            calling the slot.
        check_nargs : Optional[bool]
            If `True` and the provided `slot` requires more positional arguments than
            the signature of this Signal, raise `TypeError`. by default `True`.
        thread: Thread | Literal["main", "current"] | None
            If `None` (the default), this slot will be invoked immediately when a signal
            is emitted, from whatever thread emitted the signal. If a thread object is
            provided, then the callback will only be immediately invoked if the signal
            is emitted from that thread.  Otherwise, the callback will be added to a
            queue. **Note!**, when using the `thread` parameter, the user is responsible
            for calling `psygnal.emit_queued()` in the corresponding thread, otherwise
            the slot will never be invoked. (See note above). (The strings `"main"` and
            `"current"` are also accepted, and will be interpreted as the
            `threading.main_thread()` and `threading.current_thread()`, respectively).
        check_types : Optional[bool]
            If `True`, An additional check will be performed to make sure that types
            declared in the slot signature are compatible with the signature
            declared by this signal, by default `False`.
        unique : Union[bool, str, None]
            If `True`, returns without connecting if the slot has already been
            connected.  If the literal string "raise" is passed to `unique`, then a
            `ValueError` will be raised if the slot is already connected.
            By default `False`.
        max_args : Optional[int]
            If provided, `slot` will be called with no more more than `max_args` when
            this SignalInstance is emitted.  (regardless of how many arguments are
            emitted).
        on_ref_error : {'raise', 'warn', 'ignore'}, optional
            What to do if a weak reference cannot be created.  If 'raise', a
            ReferenceError will be raised.  If 'warn' (default), a warning will be
            issued and a strong-reference will be used. If 'ignore' a strong-reference
            will be used (silently).

        Raises
        ------
        TypeError
            If a non-callable object is provided.
        ValueError
            If the provided slot fails validation, either due to mismatched positional
            argument requirements, or failed type checking.
        ValueError
            If `unique` is `True` and `slot` has already been connected.
        """
        if check_nargs is None:
            check_nargs = self._check_nargs_on_connect
        if check_types is None:
            check_types = self._check_types_on_connect

        def _wrapper(
            slot: F,
            max_args: int | None = max_args,
            _on_ref_err: RefErrorChoice = on_ref_error,
        ) -> F:
            if not callable(slot):
                raise TypeError(f"Cannot connect to non-callable object: {slot}")

            with self._lock:
                if unique and slot in self:
                    if unique == "raise":
                        raise ValueError(
                            "Slot already connect. Use `connect(..., unique=False)` "
                            "to allow duplicate connections"
                        )
                    return slot

                slot_sig: Signature | None = None
                if check_nargs and (max_args is None):
                    slot_sig, max_args, isqt = self._check_nargs(slot, self.signature)
                    if isqt:
                        _on_ref_err = "ignore"
                if check_types:
                    slot_sig = slot_sig or signature(slot)
                    if not _parameter_types_match(slot, self.signature, slot_sig):
                        extra = f"- Slot types {slot_sig} do not match types in signal."
                        self._raise_connection_error(slot, extra)

                cb = weak_callback(
                    slot,
                    max_args=max_args,
                    finalize=self._try_discard,
                    on_ref_error=_on_ref_err,
                )
                if thread is not None:
                    cb = QueuedCallback(cb, thread=thread)
                self._append_slot(cb)
            return slot

        return _wrapper if slot is None else _wrapper(slot)

    def _append_slot(self, slot: WeakCallback) -> None:
        """Append a slot to the list of slots."""
        # implementing this as a method allows us to override/extend it in subclasses
        self._slots.append(slot)

    def _remove_slot(self, slot: Literal["all"] | int | WeakCallback) -> None:
        """Remove a slot from the list of slots."""
        # implementing this as a method allows us to override/extend it in subclasses
        if slot == "all":
            self._slots.clear()
        elif isinstance(slot, int):
            self._slots.pop(slot)
        else:
            self._slots.remove(cast("WeakCallback", slot))

    def _try_discard(self, callback: WeakCallback, missing_ok: bool = True) -> None:
        """Try to discard a callback from the list of slots.

        Parameters
        ----------
        callback : WeakCallback
            A callback to discard.
        missing_ok : bool, optional
            If `True`, do not raise an error if the callback is not found in the list.
        """
        try:
            self._remove_slot(callback)
        except ValueError:
            if not missing_ok:
                raise

    def connect_setattr(
        self,
        obj: object,
        attr: str,
        maxargs: int | None | object = _NULL,
        *,
        on_ref_error: RefErrorChoice = "warn",
    ) -> WeakCallback[None]:
        """Bind an object attribute to the emitted value of this signal.

        Equivalent to calling `self.connect(functools.partial(setattr, obj, attr))`,
        but with additional weakref safety (i.e. a strong reference to `obj` will not
        be retained). The return object can be used to
        [`disconnect()`][psygnal.SignalInstance.disconnect], (or you can use
        [`disconnect_setattr()`][psygnal.SignalInstance.disconnect_setattr]).

        Parameters
        ----------
        obj : object
            An object.
        attr : str
            The name of an attribute on `obj` that should be set to the value of this
            signal when emitted.
        maxargs : Optional[int]
            max number of positional args to accept
        on_ref_error: {'raise', 'warn', 'ignore'}, optional
            What to do if a weak reference cannot be created.  If 'raise', a
            ReferenceError will be raised.  If 'warn' (default), a warning will be
            issued and a strong-reference will be used. If 'ignore' a strong-reference
            will be used (silently).

        Returns
        -------
        Tuple
            (weakref.ref, name, callable).  Reference to the object, name of the
            attribute, and setattr closure.  Can be used to disconnect the slot.

        Raises
        ------
        ValueError
            If this is not a single-value signal
        AttributeError
            If `obj` has no attribute `attr`.

        Examples
        --------
        >>> class T:
        ...     sig = Signal(int)
        ...
        >>> class SomeObj:
        ...     x = 1
        ...
        >>> t = T()
        >>> my_obj = SomeObj()
        >>> t.sig.connect_setattr(my_obj, 'x')
        >>> t.sig.emit(5)
        >>> assert my_obj.x == 5
        """
        if maxargs is _NULL:
            warnings.warn(
                "The default value of maxargs will change from `None` to `1` in"
                "version 0.11. To silence this warning, provide an explicit value for "
                "maxargs (`None` for current behavior, `1` for future behavior).",
                FutureWarning,
                stacklevel=2,
            )
            maxargs = None

        if not hasattr(obj, attr):
            raise AttributeError(f"Object {obj} has no attribute {attr!r}")

        with self._lock:
            caller = WeakSetattr(
                obj,
                attr,
                max_args=cast("int | None", maxargs),
                finalize=self._try_discard,
                on_ref_error=on_ref_error,
            )
            self._append_slot(caller)
        return caller

    def disconnect_setattr(
        self, obj: object, attr: str, missing_ok: bool = True
    ) -> None:
        """Disconnect a previously connected attribute setter.

        Parameters
        ----------
        obj : object
            An object.
        attr : str
            The name of an attribute on `obj` that was previously used for
            `connect_setattr`.
        missing_ok : bool
            If `False` and the provided `slot` is not connected, raises `ValueError`.
            by default `True`

        Raises
        ------
        ValueError
            If `missing_ok` is `True` and no attribute setter is connected.
        """
        # sourcery skip: merge-nested-ifs, use-next
        with self._lock:
            cb = WeakSetattr(obj, attr, on_ref_error="ignore")
            self._try_discard(cb, missing_ok)

    def connect_setitem(
        self,
        obj: object,
        key: str,
        maxargs: int | None | object = _NULL,
        *,
        on_ref_error: RefErrorChoice = "warn",
    ) -> WeakCallback[None]:
        """Bind a container item (such as a dict key) to emitted value of this signal.

        Equivalent to calling `self.connect(functools.partial(obj.__setitem__, attr))`,
        but with additional weakref safety (i.e. a strong reference to `obj` will not
        be retained). The return object can be used to
        [`disconnect()`][psygnal.SignalInstance.disconnect], (or you can use
        [`disconnect_setitem()`][psygnal.SignalInstance.disconnect_setitem]).

        Parameters
        ----------
        obj : object
            An object.
        key : str
            Name of the key in `obj` that should be set to the value of this
            signal when emitted
        maxargs : Optional[int]
            max number of positional args to accept
        on_ref_error: {'raise', 'warn', 'ignore'}, optional
            What to do if a weak reference cannot be created.  If 'raise', a
            ReferenceError will be raised.  If 'warn' (default), a warning will be
            issued and a strong-reference will be used. If 'ignore' a strong-reference
            will be used (silently).

        Returns
        -------
        Tuple
            (weakref.ref, name, callable).  Reference to the object, name of the
            attribute, and setitem closure.  Can be used to disconnect the slot.

        Raises
        ------
        ValueError
            If this is not a single-value signal
        TypeError
            If `obj` does not support __setitem__.

        Examples
        --------
        >>> class T:
        ...     sig = Signal(int)
        ...
        >>> t = T()
        >>> my_obj = dict()
        >>> t.sig.connect_setitem(my_obj, 'x')
        >>> t.sig.emit(5)
        >>> assert my_obj == {'x': 5}
        """
        if maxargs is _NULL:
            warnings.warn(
                "The default value of maxargs will change from `None` to `1` in"
                "version 0.11. To silence this warning, provide an explicit value for "
                "maxargs (`None` for current behavior, `1` for future behavior).",
                FutureWarning,
                stacklevel=2,
            )
            maxargs = None

        if not hasattr(obj, "__setitem__"):
            raise TypeError(f"Object {obj} does not support __setitem__")

        with self._lock:
            caller = WeakSetitem(
                obj,
                key,
                max_args=cast("int | None", maxargs),
                finalize=self._try_discard,
                on_ref_error=on_ref_error,
            )
            self._append_slot(caller)

        return caller

    def disconnect_setitem(
        self, obj: object, key: str, missing_ok: bool = True
    ) -> None:
        """Disconnect a previously connected item setter.

        Parameters
        ----------
        obj : object
            An object.
        key : str
            The name of a key in `obj` that was previously used for
            `connect_setitem`.
        missing_ok : bool
            If `False` and the provided `slot` is not connected, raises `ValueError`.
            by default `True`

        Raises
        ------
        ValueError
            If `missing_ok` is `True` and no item setter is connected.
        """
        if not hasattr(obj, "__setitem__"):
            raise TypeError(f"Object {obj} does not support __setitem__")

        # sourcery skip: merge-nested-ifs, use-next
        with self._lock:
            caller = WeakSetitem(obj, key, on_ref_error="ignore")
            self._try_discard(caller, missing_ok)

    def _check_nargs(
        self, slot: Callable, spec: Signature
    ) -> tuple[Signature | None, int | None, bool]:
        """Make sure slot is compatible with signature.

        Also returns the maximum number of arguments that we can pass to the slot

        Returns
        -------
        slot_sig : Signature | None
            The signature of the slot, or None if it could not be determined.
        maxargs : int | None
            The maximum number of arguments that we can pass to the slot.
        is_qt : bool
            Whether the slot is a Qt slot.
        """
        try:
            slot_sig = _get_signature_possibly_qt(slot)
        except ValueError as e:
            warnings.warn(
                f"{e}. To silence this warning, connect with " "`check_nargs=False`",
                stacklevel=2,
            )
            return None, None, False
        try:
            minargs, maxargs = _acceptable_posarg_range(slot_sig)
        except ValueError as e:
            if isinstance(slot, partial):
                raise ValueError(
                    f"{e}. (Note: prefer using positional args with "
                    "functools.partials when possible)."
                ) from e
            raise

        # if `slot` requires more arguments than we will provide, raise.
        if minargs > (n_spec_params := len(spec.parameters)):
            extra = (
                f"- Slot requires at least {minargs} positional "
                f"arguments, but spec only provides {n_spec_params}"
            )
            self._raise_connection_error(slot, extra)

        return None if isinstance(slot_sig, str) else slot_sig, maxargs, True

    def _raise_connection_error(self, slot: Callable, extra: str = "") -> NoReturn:
        name = getattr(slot, "__name__", str(slot))
        msg = f"Cannot connect slot {name!r} with signature: {signature(slot)}:\n"
        msg += extra
        msg += f"\n\nAccepted signature: {self.signature}"
        raise ValueError(msg)

    def _slot_index(self, slot: Callable) -> int:
        """Get index of `slot` in `self._slots`.  Return -1 if not connected."""
        with self._lock:
            normed = weak_callback(slot, on_ref_error="ignore")
            # NOTE:
            # the == method here relies on the __eq__ method of each SlotCaller subclass
            return next((i for i, s in enumerate(self._slots) if s == normed), -1)

    def disconnect(self, slot: Callable | None = None, missing_ok: bool = True) -> None:
        """Disconnect slot from signal.

        Parameters
        ----------
        slot : callable, optional
            The specific slot to disconnect.  If `None`, all slots will be disconnected,
            by default `None`
        missing_ok : Optional[bool]
            If `False` and the provided `slot` is not connected, raises `ValueError.
            by default `True`

        Raises
        ------
        ValueError
            If `slot` is not connected and `missing_ok` is False.
        """
        with self._lock:
            if slot is None:
                # NOTE: clearing an empty list is actually a RuntimeError in Qt
                self._remove_slot("all")
                return

            idx = self._slot_index(slot)
            if idx != -1:
                self._remove_slot(idx)
            elif not missing_ok:
                raise ValueError(f"slot is not connected: {slot}")

    def __contains__(self, slot: Callable) -> bool:
        """Return `True` if slot is connected."""
        return self._slot_index(slot) >= 0

    def __len__(self) -> int:
        """Return number of connected slots."""
        return len(self._slots)

    def emit(
        self, *args: Any, check_nargs: bool = False, check_types: bool = False
    ) -> None:
        """Emit this signal with arguments `args`.

        !!! note

            `check_args` and `check_types` both add overhead when calling emit.

        Parameters
        ----------
        *args : Any
            These arguments will be passed when calling each slot (unless the slot
            accepts fewer arguments, in which case extra args will be discarded.)
        check_nargs : bool
            If `False` and the provided arguments cannot be successfully bound to the
            signature of this Signal, raise `TypeError`.  Incurs some overhead.
            by default False.
        check_types : bool
            If `False` and the provided arguments do not match the types declared by
            the signature of this Signal, raise `TypeError`.  Incurs some overhead.
            by default False.

        Raises
        ------
        TypeError
            If `check_nargs` and/or `check_types` are `True`, and the corresponding
            checks fail.
        """
        if self._is_blocked:
            return None

        if check_nargs:
            try:
                self.signature.bind(*args)
            except TypeError as e:
                raise TypeError(
                    f"Cannot emit args {args} from signal {self!r} with "
                    f"signature {self.signature}:\n{e}"
                ) from e

        if check_types and not _parameter_types_match(
            lambda: None, self.signature, _build_signature(*[type(a) for a in args])
        ):
            raise TypeError(
                f"Types provided to '{self.name}.emit' "
                f"{tuple(type(a).__name__ for a in args)} do not match signal "
                f"signature: {self.signature}"
            )

        if self._is_paused:
            self._args_queue.append(args)
            return None

        if SignalInstance._debug_hook is not None:
            from ._group import EmissionInfo

            SignalInstance._debug_hook(EmissionInfo(self, args))

        self._run_emit_loop(args)
        return None

    def __call__(
        self, *args: Any, check_nargs: bool = False, check_types: bool = False
    ) -> None:
        """Alias for `emit()`."""
        return self.emit(
            *args,
            check_nargs=check_nargs,
            check_types=check_types,
        )

    def _run_emit_loop(self, args: tuple[Any, ...]) -> None:
        # allow receiver to query sender with Signal.current_emitter()
        with self._lock:
            with Signal._emitting(self):
                for caller in self._slots:
                    try:
                        caller.cb(args)
                    except Exception as e:
                        raise EmitLoopError(
                            cb=caller, args=args, exc=e, signal=self
                        ) from e

        return None

    def block(self, exclude: Iterable[str | SignalInstance] = ()) -> None:
        """Block this signal from emitting.

        NOTE: the `exclude` argument is only for SignalGroup subclass, but we
        have to include it here to make mypyc happy.
        """
        self._is_blocked = True

    def unblock(self) -> None:
        """Unblock this signal, allowing it to emit."""
        self._is_blocked = False

    def blocked(self) -> ContextManager[None]:
        """Context manager to temporarily block this signal.

        Useful if you need to temporarily block all emission of a given signal,
        (for example, to avoid a recursive signal loop)

        Examples
        --------
        ```python
        class MyEmitter:
            changed = Signal()

            def make_a_change(self):
                self.changed.emit()

        obj = MyEmitter()

        with obj.changed.blocked()
            obj.make_a_change()  # will NOT emit a changed signal.
        ```
        """
        return _SignalBlocker(self)

    def pause(self) -> None:
        """Pause all emission and collect *args tuples from emit().

        args passed to `emit` will be collected and re-emitted when `resume()` is
        called. For a context manager version, see `paused()`.
        """
        self._is_paused = True

    def resume(self, reducer: ReducerFunc | None = None, initial: Any = _NULL) -> None:
        """Resume (unpause) this signal, emitting everything in the queue.

        Parameters
        ----------
        reducer : Callable | None
            A optional function to reduce the args collected while paused into a single
            emitted group of args.  If not provided, all emissions will be re-emitted
            as they were collected when the signal is resumed. May be:

            - a function that takes two args tuples and returns a single args tuple.
              This will be passed to `functools.reduce` and is expected to reduce all
              collected/emitted args into a single tuple.
              For example, three `emit(1)` events would be reduced and re-emitted as
              follows: `self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))`
            - a function that takes a single argument (an iterable of args tuples) and
              returns a tuple (the reduced args). This will be *not* be passed to
              `functools.reduce`. If `reducer` is a function that takes a single
              argument, `initial` will be ignored.
        initial: any, optional
            initial value to pass to `functools.reduce`

        Examples
        --------
        >>> class T:
        ...     sig = Signal(int)
        >>> t = T()
        >>> t.sig.pause()
        >>> t.sig.emit(1)
        >>> t.sig.emit(2)
        >>> t.sig.emit(3)
        >>> t.sig.resume(lambda a, b: (a[0].union(set(b)),), (set(),))
        >>> # results in t.sig.emit({1, 2, 3})
        """
        self._is_paused = False
        # not sure why this attribute wouldn't be set, but when resuming in
        # EventedModel.update, it may be undefined (as seen in tests)
        if not getattr(self, "_args_queue", None):
            return
        if len(self._slots) == 0:
            self._args_queue.clear()
            return

        if reducer is not None:
            if len(inspect.signature(reducer).parameters) == 1:
                args = cast("ReducerOneArg", reducer)(self._args_queue)
            else:
                reducer = cast("ReducerTwoArgs", reducer)
                if initial is _NULL:
                    args = reduce(reducer, self._args_queue)
                else:
                    args = reduce(reducer, self._args_queue, initial)
            self._run_emit_loop(args)
        else:
            for args in self._args_queue:
                self._run_emit_loop(args)
        self._args_queue.clear()

    def paused(
        self, reducer: ReducerFunc | None = None, initial: Any = _NULL
    ) -> ContextManager[None]:
        """Context manager to temporarily pause this signal.

        Parameters
        ----------
        reducer : Callable | None
            A optional function to reduce the args collected while paused into a single
            emitted group of args.  If not provided, all emissions will be re-emitted
            as they were collected when the signal is resumed. May be:

            - a function that takes two args tuples and returns a single args tuple.
              This will be passed to `functools.reduce` and is expected to reduce all
              collected/emitted args into a single tuple.
              For example, three `emit(1)` events would be reduced and re-emitted as
              follows: `self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))`
            - a function that takes a single argument (an iterable of args tuples) and
              returns a tuple (the reduced args). This will be *not* be passed to
              `functools.reduce`. If `reducer` is a function that takes a single
              argument, `initial` will be ignored.
        initial: any, optional
            initial value to pass to `functools.reduce`

        Examples
        --------
        >>> with obj.signal.paused(lambda a, b: (a[0].union(set(b)),), (set(),)):
        ...     t.sig.emit(1)
        ...     t.sig.emit(2)
        ...     t.sig.emit(3)
        >>> # results in obj.signal.emit({1, 2, 3})
        """
        return _SignalPauser(self, reducer, initial)

    def __getstate__(self) -> dict:
        """Return dict of current state, for pickle."""
        attrs = (
            "_signature",
            "_name",
            "_is_blocked",
            "_is_paused",
            "_args_queue",
            "_check_nargs_on_connect",
            "_check_types_on_connect",
        )
        dd = {slot: getattr(self, slot) for slot in attrs}
        dd["_instance"] = self._instance()
        dd["_slots"] = [x for x in self._slots if isinstance(x, StrongFunction)]
        if len(self._slots) > len(dd["_slots"]):
            warnings.warn(
                "Pickling a SignalInstance does not copy connected weakly referenced "
                "slots.",
                stacklevel=2,
            )

        return dd

    def __setstate__(self, state: dict) -> None:
        """Restore state from pickle."""
        # don't use __dict__, mypyc doesn't have it
        for k, v in state.items():
            if k == "_instance":
                self._instance = self._instance_ref(v)
            else:
                setattr(self, k, v)
        self._lock = threading.RLock()

instance: Any property #

Object that emits this SignalInstance.

name: str property #

Name of this SignalInstance.

signature: Signature property #

Signature supported by this SignalInstance.

block(exclude=()) #

Block this signal from emitting.

NOTE: the exclude argument is only for SignalGroup subclass, but we have to include it here to make mypyc happy.

Source code in psygnal/_signal.py
961
962
963
964
965
966
967
def block(self, exclude: Iterable[str | SignalInstance] = ()) -> None:
    """Block this signal from emitting.

    NOTE: the `exclude` argument is only for SignalGroup subclass, but we
    have to include it here to make mypyc happy.
    """
    self._is_blocked = True

blocked() #

Context manager to temporarily block this signal.

Useful if you need to temporarily block all emission of a given signal, (for example, to avoid a recursive signal loop)

Examples:

class MyEmitter:
    changed = Signal()

    def make_a_change(self):
        self.changed.emit()

obj = MyEmitter()

with obj.changed.blocked()
    obj.make_a_change()  # will NOT emit a changed signal.
Source code in psygnal/_signal.py
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
def blocked(self) -> ContextManager[None]:
    """Context manager to temporarily block this signal.

    Useful if you need to temporarily block all emission of a given signal,
    (for example, to avoid a recursive signal loop)

    Examples
    --------
    ```python
    class MyEmitter:
        changed = Signal()

        def make_a_change(self):
            self.changed.emit()

    obj = MyEmitter()

    with obj.changed.blocked()
        obj.make_a_change()  # will NOT emit a changed signal.
    ```
    """
    return _SignalBlocker(self)

connect(slot=None, *, thread=None, check_nargs=None, check_types=None, unique=False, max_args=None, on_ref_error='warn') #

Connect a callback (slot) to this signal.

slot is compatible if:

  • it requires no more than the number of positional arguments emitted by this SignalInstance. (It may require less)
  • it has no required keyword arguments (keyword only arguments that have no default).
  • if check_types is True, the parameter types in the callback signature must match the signature of this SignalInstance.

This method may be used as a decorator.

@signal.connect
def my_function():
    ...

Important

If a signal is connected with thread != None, then it is up to the user to ensure that psygnal.emit_queued is called, or that one of the backend convenience functions is used (e.g. psygnal.qt.start_emitting_from_queue). Otherwise, callbacks that are connected to signals that are emitted from another thread will never be called.

Parameters:

  • slot (Callable) –

    A callable to connect to this signal. If the callable accepts less arguments than the signature of this slot, then they will be discarded when calling the slot.

  • check_nargs (Optional[bool]) –

    If True and the provided slot requires more positional arguments than the signature of this Signal, raise TypeError. by default True.

  • thread (threading.Thread | Literal['main', 'current'] | None) –

    If None (the default), this slot will be invoked immediately when a signal is emitted, from whatever thread emitted the signal. If a thread object is provided, then the callback will only be immediately invoked if the signal is emitted from that thread. Otherwise, the callback will be added to a queue. Note!, when using the thread parameter, the user is responsible for calling psygnal.emit_queued() in the corresponding thread, otherwise the slot will never be invoked. (See note above). (The strings "main" and "current" are also accepted, and will be interpreted as the threading.main_thread() and threading.current_thread(), respectively).

  • check_types (Optional[bool]) –

    If True, An additional check will be performed to make sure that types declared in the slot signature are compatible with the signature declared by this signal, by default False.

  • unique (Union[bool, str, None]) –

    If True, returns without connecting if the slot has already been connected. If the literal string "raise" is passed to unique, then a ValueError will be raised if the slot is already connected. By default False.

  • max_args (Optional[int]) –

    If provided, slot will be called with no more more than max_args when this SignalInstance is emitted. (regardless of how many arguments are emitted).

  • on_ref_error (RefErrorChoice) –

    What to do if a weak reference cannot be created. If 'raise', a ReferenceError will be raised. If 'warn' (default), a warning will be issued and a strong-reference will be used. If 'ignore' a strong-reference will be used (silently).

Raises:

  • TypeError

    If a non-callable object is provided.

  • ValueError

    If the provided slot fails validation, either due to mismatched positional argument requirements, or failed type checking.

  • ValueError

    If unique is True and slot has already been connected.

Source code in psygnal/_signal.py
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
def connect(
    self,
    slot: F | None = None,
    *,
    thread: threading.Thread | Literal["main", "current"] | None = None,
    check_nargs: bool | None = None,
    check_types: bool | None = None,
    unique: bool | str = False,
    max_args: int | None = None,
    on_ref_error: RefErrorChoice = "warn",
) -> Callable[[F], F] | F:
    """Connect a callback (`slot`) to this signal.

    `slot` is compatible if:

    * it requires no more than the number of positional arguments emitted by this
      `SignalInstance`.  (It *may* require less)
    * it has no *required* keyword arguments (keyword only arguments that have
      no default).
    * if `check_types` is `True`, the parameter types in the callback signature must
      match the signature of this `SignalInstance`.

    This method may be used as a decorator.

    ```python
    @signal.connect
    def my_function():
        ...
    ```

    !!!important
        If a signal is connected with `thread != None`, then it is up to the user
        to ensure that `psygnal.emit_queued` is called, or that one of the backend
        convenience functions is used (e.g. `psygnal.qt.start_emitting_from_queue`).
        Otherwise, callbacks that are connected to signals that are emitted from
        another thread will never be called.

    Parameters
    ----------
    slot : Callable
        A callable to connect to this signal.  If the callable accepts less
        arguments than the signature of this slot, then they will be discarded when
        calling the slot.
    check_nargs : Optional[bool]
        If `True` and the provided `slot` requires more positional arguments than
        the signature of this Signal, raise `TypeError`. by default `True`.
    thread: Thread | Literal["main", "current"] | None
        If `None` (the default), this slot will be invoked immediately when a signal
        is emitted, from whatever thread emitted the signal. If a thread object is
        provided, then the callback will only be immediately invoked if the signal
        is emitted from that thread.  Otherwise, the callback will be added to a
        queue. **Note!**, when using the `thread` parameter, the user is responsible
        for calling `psygnal.emit_queued()` in the corresponding thread, otherwise
        the slot will never be invoked. (See note above). (The strings `"main"` and
        `"current"` are also accepted, and will be interpreted as the
        `threading.main_thread()` and `threading.current_thread()`, respectively).
    check_types : Optional[bool]
        If `True`, An additional check will be performed to make sure that types
        declared in the slot signature are compatible with the signature
        declared by this signal, by default `False`.
    unique : Union[bool, str, None]
        If `True`, returns without connecting if the slot has already been
        connected.  If the literal string "raise" is passed to `unique`, then a
        `ValueError` will be raised if the slot is already connected.
        By default `False`.
    max_args : Optional[int]
        If provided, `slot` will be called with no more more than `max_args` when
        this SignalInstance is emitted.  (regardless of how many arguments are
        emitted).
    on_ref_error : {'raise', 'warn', 'ignore'}, optional
        What to do if a weak reference cannot be created.  If 'raise', a
        ReferenceError will be raised.  If 'warn' (default), a warning will be
        issued and a strong-reference will be used. If 'ignore' a strong-reference
        will be used (silently).

    Raises
    ------
    TypeError
        If a non-callable object is provided.
    ValueError
        If the provided slot fails validation, either due to mismatched positional
        argument requirements, or failed type checking.
    ValueError
        If `unique` is `True` and `slot` has already been connected.
    """
    if check_nargs is None:
        check_nargs = self._check_nargs_on_connect
    if check_types is None:
        check_types = self._check_types_on_connect

    def _wrapper(
        slot: F,
        max_args: int | None = max_args,
        _on_ref_err: RefErrorChoice = on_ref_error,
    ) -> F:
        if not callable(slot):
            raise TypeError(f"Cannot connect to non-callable object: {slot}")

        with self._lock:
            if unique and slot in self:
                if unique == "raise":
                    raise ValueError(
                        "Slot already connect. Use `connect(..., unique=False)` "
                        "to allow duplicate connections"
                    )
                return slot

            slot_sig: Signature | None = None
            if check_nargs and (max_args is None):
                slot_sig, max_args, isqt = self._check_nargs(slot, self.signature)
                if isqt:
                    _on_ref_err = "ignore"
            if check_types:
                slot_sig = slot_sig or signature(slot)
                if not _parameter_types_match(slot, self.signature, slot_sig):
                    extra = f"- Slot types {slot_sig} do not match types in signal."
                    self._raise_connection_error(slot, extra)

            cb = weak_callback(
                slot,
                max_args=max_args,
                finalize=self._try_discard,
                on_ref_error=_on_ref_err,
            )
            if thread is not None:
                cb = QueuedCallback(cb, thread=thread)
            self._append_slot(cb)
        return slot

    return _wrapper if slot is None else _wrapper(slot)

connect_setattr(obj, attr, maxargs=_NULL, *, on_ref_error='warn') #

Bind an object attribute to the emitted value of this signal.

Equivalent to calling self.connect(functools.partial(setattr, obj, attr)), but with additional weakref safety (i.e. a strong reference to obj will not be retained). The return object can be used to disconnect(), (or you can use disconnect_setattr()).

Parameters:

  • obj (object) –

    An object.

  • attr (str) –

    The name of an attribute on obj that should be set to the value of this signal when emitted.

  • maxargs (Optional[int]) –

    max number of positional args to accept

  • on_ref_error (RefErrorChoice) –

    What to do if a weak reference cannot be created. If 'raise', a ReferenceError will be raised. If 'warn' (default), a warning will be issued and a strong-reference will be used. If 'ignore' a strong-reference will be used (silently).

Returns:

  • Tuple

    (weakref.ref, name, callable). Reference to the object, name of the attribute, and setattr closure. Can be used to disconnect the slot.

Raises:

Examples:

>>> class T:
...     sig = Signal(int)
...
>>> class SomeObj:
...     x = 1
...
>>> t = T()
>>> my_obj = SomeObj()
>>> t.sig.connect_setattr(my_obj, 'x')
>>> t.sig.emit(5)
>>> assert my_obj.x == 5
Source code in psygnal/_signal.py
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
def connect_setattr(
    self,
    obj: object,
    attr: str,
    maxargs: int | None | object = _NULL,
    *,
    on_ref_error: RefErrorChoice = "warn",
) -> WeakCallback[None]:
    """Bind an object attribute to the emitted value of this signal.

    Equivalent to calling `self.connect(functools.partial(setattr, obj, attr))`,
    but with additional weakref safety (i.e. a strong reference to `obj` will not
    be retained). The return object can be used to
    [`disconnect()`][psygnal.SignalInstance.disconnect], (or you can use
    [`disconnect_setattr()`][psygnal.SignalInstance.disconnect_setattr]).

    Parameters
    ----------
    obj : object
        An object.
    attr : str
        The name of an attribute on `obj` that should be set to the value of this
        signal when emitted.
    maxargs : Optional[int]
        max number of positional args to accept
    on_ref_error: {'raise', 'warn', 'ignore'}, optional
        What to do if a weak reference cannot be created.  If 'raise', a
        ReferenceError will be raised.  If 'warn' (default), a warning will be
        issued and a strong-reference will be used. If 'ignore' a strong-reference
        will be used (silently).

    Returns
    -------
    Tuple
        (weakref.ref, name, callable).  Reference to the object, name of the
        attribute, and setattr closure.  Can be used to disconnect the slot.

    Raises
    ------
    ValueError
        If this is not a single-value signal
    AttributeError
        If `obj` has no attribute `attr`.

    Examples
    --------
    >>> class T:
    ...     sig = Signal(int)
    ...
    >>> class SomeObj:
    ...     x = 1
    ...
    >>> t = T()
    >>> my_obj = SomeObj()
    >>> t.sig.connect_setattr(my_obj, 'x')
    >>> t.sig.emit(5)
    >>> assert my_obj.x == 5
    """
    if maxargs is _NULL:
        warnings.warn(
            "The default value of maxargs will change from `None` to `1` in"
            "version 0.11. To silence this warning, provide an explicit value for "
            "maxargs (`None` for current behavior, `1` for future behavior).",
            FutureWarning,
            stacklevel=2,
        )
        maxargs = None

    if not hasattr(obj, attr):
        raise AttributeError(f"Object {obj} has no attribute {attr!r}")

    with self._lock:
        caller = WeakSetattr(
            obj,
            attr,
            max_args=cast("int | None", maxargs),
            finalize=self._try_discard,
            on_ref_error=on_ref_error,
        )
        self._append_slot(caller)
    return caller

connect_setitem(obj, key, maxargs=_NULL, *, on_ref_error='warn') #

Bind a container item (such as a dict key) to emitted value of this signal.

Equivalent to calling self.connect(functools.partial(obj.__setitem__, attr)), but with additional weakref safety (i.e. a strong reference to obj will not be retained). The return object can be used to disconnect(), (or you can use disconnect_setitem()).

Parameters:

  • obj (object) –

    An object.

  • key (str) –

    Name of the key in obj that should be set to the value of this signal when emitted

  • maxargs (Optional[int]) –

    max number of positional args to accept

  • on_ref_error (RefErrorChoice) –

    What to do if a weak reference cannot be created. If 'raise', a ReferenceError will be raised. If 'warn' (default), a warning will be issued and a strong-reference will be used. If 'ignore' a strong-reference will be used (silently).

Returns:

  • Tuple

    (weakref.ref, name, callable). Reference to the object, name of the attribute, and setitem closure. Can be used to disconnect the slot.

Raises:

  • ValueError

    If this is not a single-value signal

  • TypeError

    If obj does not support setitem.

Examples:

>>> class T:
...     sig = Signal(int)
...
>>> t = T()
>>> my_obj = dict()
>>> t.sig.connect_setitem(my_obj, 'x')
>>> t.sig.emit(5)
>>> assert my_obj == {'x': 5}
Source code in psygnal/_signal.py
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
def connect_setitem(
    self,
    obj: object,
    key: str,
    maxargs: int | None | object = _NULL,
    *,
    on_ref_error: RefErrorChoice = "warn",
) -> WeakCallback[None]:
    """Bind a container item (such as a dict key) to emitted value of this signal.

    Equivalent to calling `self.connect(functools.partial(obj.__setitem__, attr))`,
    but with additional weakref safety (i.e. a strong reference to `obj` will not
    be retained). The return object can be used to
    [`disconnect()`][psygnal.SignalInstance.disconnect], (or you can use
    [`disconnect_setitem()`][psygnal.SignalInstance.disconnect_setitem]).

    Parameters
    ----------
    obj : object
        An object.
    key : str
        Name of the key in `obj` that should be set to the value of this
        signal when emitted
    maxargs : Optional[int]
        max number of positional args to accept
    on_ref_error: {'raise', 'warn', 'ignore'}, optional
        What to do if a weak reference cannot be created.  If 'raise', a
        ReferenceError will be raised.  If 'warn' (default), a warning will be
        issued and a strong-reference will be used. If 'ignore' a strong-reference
        will be used (silently).

    Returns
    -------
    Tuple
        (weakref.ref, name, callable).  Reference to the object, name of the
        attribute, and setitem closure.  Can be used to disconnect the slot.

    Raises
    ------
    ValueError
        If this is not a single-value signal
    TypeError
        If `obj` does not support __setitem__.

    Examples
    --------
    >>> class T:
    ...     sig = Signal(int)
    ...
    >>> t = T()
    >>> my_obj = dict()
    >>> t.sig.connect_setitem(my_obj, 'x')
    >>> t.sig.emit(5)
    >>> assert my_obj == {'x': 5}
    """
    if maxargs is _NULL:
        warnings.warn(
            "The default value of maxargs will change from `None` to `1` in"
            "version 0.11. To silence this warning, provide an explicit value for "
            "maxargs (`None` for current behavior, `1` for future behavior).",
            FutureWarning,
            stacklevel=2,
        )
        maxargs = None

    if not hasattr(obj, "__setitem__"):
        raise TypeError(f"Object {obj} does not support __setitem__")

    with self._lock:
        caller = WeakSetitem(
            obj,
            key,
            max_args=cast("int | None", maxargs),
            finalize=self._try_discard,
            on_ref_error=on_ref_error,
        )
        self._append_slot(caller)

    return caller

disconnect(slot=None, missing_ok=True) #

Disconnect slot from signal.

Parameters:

  • slot (callable, optional) –

    The specific slot to disconnect. If None, all slots will be disconnected, by default None

  • missing_ok (Optional[bool]) –

    If False and the provided slot is not connected, raises ValueError. by defaultTrue`

Raises:

  • ValueError

    If slot is not connected and missing_ok is False.

Source code in psygnal/_signal.py
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
def disconnect(self, slot: Callable | None = None, missing_ok: bool = True) -> None:
    """Disconnect slot from signal.

    Parameters
    ----------
    slot : callable, optional
        The specific slot to disconnect.  If `None`, all slots will be disconnected,
        by default `None`
    missing_ok : Optional[bool]
        If `False` and the provided `slot` is not connected, raises `ValueError.
        by default `True`

    Raises
    ------
    ValueError
        If `slot` is not connected and `missing_ok` is False.
    """
    with self._lock:
        if slot is None:
            # NOTE: clearing an empty list is actually a RuntimeError in Qt
            self._remove_slot("all")
            return

        idx = self._slot_index(slot)
        if idx != -1:
            self._remove_slot(idx)
        elif not missing_ok:
            raise ValueError(f"slot is not connected: {slot}")

disconnect_setattr(obj, attr, missing_ok=True) #

Disconnect a previously connected attribute setter.

Parameters:

  • obj (object) –

    An object.

  • attr (str) –

    The name of an attribute on obj that was previously used for connect_setattr.

  • missing_ok (bool) –

    If False and the provided slot is not connected, raises ValueError. by default True

Raises:

  • ValueError

    If missing_ok is True and no attribute setter is connected.

Source code in psygnal/_signal.py
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
def disconnect_setattr(
    self, obj: object, attr: str, missing_ok: bool = True
) -> None:
    """Disconnect a previously connected attribute setter.

    Parameters
    ----------
    obj : object
        An object.
    attr : str
        The name of an attribute on `obj` that was previously used for
        `connect_setattr`.
    missing_ok : bool
        If `False` and the provided `slot` is not connected, raises `ValueError`.
        by default `True`

    Raises
    ------
    ValueError
        If `missing_ok` is `True` and no attribute setter is connected.
    """
    # sourcery skip: merge-nested-ifs, use-next
    with self._lock:
        cb = WeakSetattr(obj, attr, on_ref_error="ignore")
        self._try_discard(cb, missing_ok)

disconnect_setitem(obj, key, missing_ok=True) #

Disconnect a previously connected item setter.

Parameters:

  • obj (object) –

    An object.

  • key (str) –

    The name of a key in obj that was previously used for connect_setitem.

  • missing_ok (bool) –

    If False and the provided slot is not connected, raises ValueError. by default True

Raises:

  • ValueError

    If missing_ok is True and no item setter is connected.

Source code in psygnal/_signal.py
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
def disconnect_setitem(
    self, obj: object, key: str, missing_ok: bool = True
) -> None:
    """Disconnect a previously connected item setter.

    Parameters
    ----------
    obj : object
        An object.
    key : str
        The name of a key in `obj` that was previously used for
        `connect_setitem`.
    missing_ok : bool
        If `False` and the provided `slot` is not connected, raises `ValueError`.
        by default `True`

    Raises
    ------
    ValueError
        If `missing_ok` is `True` and no item setter is connected.
    """
    if not hasattr(obj, "__setitem__"):
        raise TypeError(f"Object {obj} does not support __setitem__")

    # sourcery skip: merge-nested-ifs, use-next
    with self._lock:
        caller = WeakSetitem(obj, key, on_ref_error="ignore")
        self._try_discard(caller, missing_ok)

emit(*args, check_nargs=False, check_types=False) #

Emit this signal with arguments args.

Note

check_args and check_types both add overhead when calling emit.

Parameters:

  • *args (Any) –

    These arguments will be passed when calling each slot (unless the slot accepts fewer arguments, in which case extra args will be discarded.)

  • check_nargs (bool) –

    If False and the provided arguments cannot be successfully bound to the signature of this Signal, raise TypeError. Incurs some overhead. by default False.

  • check_types (bool) –

    If False and the provided arguments do not match the types declared by the signature of this Signal, raise TypeError. Incurs some overhead. by default False.

Raises:

  • TypeError

    If check_nargs and/or check_types are True, and the corresponding checks fail.

Source code in psygnal/_signal.py
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
def emit(
    self, *args: Any, check_nargs: bool = False, check_types: bool = False
) -> None:
    """Emit this signal with arguments `args`.

    !!! note

        `check_args` and `check_types` both add overhead when calling emit.

    Parameters
    ----------
    *args : Any
        These arguments will be passed when calling each slot (unless the slot
        accepts fewer arguments, in which case extra args will be discarded.)
    check_nargs : bool
        If `False` and the provided arguments cannot be successfully bound to the
        signature of this Signal, raise `TypeError`.  Incurs some overhead.
        by default False.
    check_types : bool
        If `False` and the provided arguments do not match the types declared by
        the signature of this Signal, raise `TypeError`.  Incurs some overhead.
        by default False.

    Raises
    ------
    TypeError
        If `check_nargs` and/or `check_types` are `True`, and the corresponding
        checks fail.
    """
    if self._is_blocked:
        return None

    if check_nargs:
        try:
            self.signature.bind(*args)
        except TypeError as e:
            raise TypeError(
                f"Cannot emit args {args} from signal {self!r} with "
                f"signature {self.signature}:\n{e}"
            ) from e

    if check_types and not _parameter_types_match(
        lambda: None, self.signature, _build_signature(*[type(a) for a in args])
    ):
        raise TypeError(
            f"Types provided to '{self.name}.emit' "
            f"{tuple(type(a).__name__ for a in args)} do not match signal "
            f"signature: {self.signature}"
        )

    if self._is_paused:
        self._args_queue.append(args)
        return None

    if SignalInstance._debug_hook is not None:
        from ._group import EmissionInfo

        SignalInstance._debug_hook(EmissionInfo(self, args))

    self._run_emit_loop(args)
    return None

pause() #

Pause all emission and collect *args tuples from emit().

args passed to emit will be collected and re-emitted when resume() is called. For a context manager version, see paused().

Source code in psygnal/_signal.py
 996
 997
 998
 999
1000
1001
1002
def pause(self) -> None:
    """Pause all emission and collect *args tuples from emit().

    args passed to `emit` will be collected and re-emitted when `resume()` is
    called. For a context manager version, see `paused()`.
    """
    self._is_paused = True

paused(reducer=None, initial=_NULL) #

Context manager to temporarily pause this signal.

Parameters:

  • reducer (Callable | None) –

    A optional function to reduce the args collected while paused into a single emitted group of args. If not provided, all emissions will be re-emitted as they were collected when the signal is resumed. May be:

    • a function that takes two args tuples and returns a single args tuple. This will be passed to functools.reduce and is expected to reduce all collected/emitted args into a single tuple. For example, three emit(1) events would be reduced and re-emitted as follows: self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))
    • a function that takes a single argument (an iterable of args tuples) and returns a tuple (the reduced args). This will be not be passed to functools.reduce. If reducer is a function that takes a single argument, initial will be ignored.
  • initial (Any) –

    initial value to pass to functools.reduce

Examples:

>>> with obj.signal.paused(lambda a, b: (a[0].union(set(b)),), (set(),)):
...     t.sig.emit(1)
...     t.sig.emit(2)
...     t.sig.emit(3)
>>> # results in obj.signal.emit({1, 2, 3})
Source code in psygnal/_signal.py
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
def paused(
    self, reducer: ReducerFunc | None = None, initial: Any = _NULL
) -> ContextManager[None]:
    """Context manager to temporarily pause this signal.

    Parameters
    ----------
    reducer : Callable | None
        A optional function to reduce the args collected while paused into a single
        emitted group of args.  If not provided, all emissions will be re-emitted
        as they were collected when the signal is resumed. May be:

        - a function that takes two args tuples and returns a single args tuple.
          This will be passed to `functools.reduce` and is expected to reduce all
          collected/emitted args into a single tuple.
          For example, three `emit(1)` events would be reduced and re-emitted as
          follows: `self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))`
        - a function that takes a single argument (an iterable of args tuples) and
          returns a tuple (the reduced args). This will be *not* be passed to
          `functools.reduce`. If `reducer` is a function that takes a single
          argument, `initial` will be ignored.
    initial: any, optional
        initial value to pass to `functools.reduce`

    Examples
    --------
    >>> with obj.signal.paused(lambda a, b: (a[0].union(set(b)),), (set(),)):
    ...     t.sig.emit(1)
    ...     t.sig.emit(2)
    ...     t.sig.emit(3)
    >>> # results in obj.signal.emit({1, 2, 3})
    """
    return _SignalPauser(self, reducer, initial)

resume(reducer=None, initial=_NULL) #

Resume (unpause) this signal, emitting everything in the queue.

Parameters:

  • reducer (Callable | None) –

    A optional function to reduce the args collected while paused into a single emitted group of args. If not provided, all emissions will be re-emitted as they were collected when the signal is resumed. May be:

    • a function that takes two args tuples and returns a single args tuple. This will be passed to functools.reduce and is expected to reduce all collected/emitted args into a single tuple. For example, three emit(1) events would be reduced and re-emitted as follows: self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))
    • a function that takes a single argument (an iterable of args tuples) and returns a tuple (the reduced args). This will be not be passed to functools.reduce. If reducer is a function that takes a single argument, initial will be ignored.
  • initial (Any) –

    initial value to pass to functools.reduce

Examples:

>>> class T:
...     sig = Signal(int)
>>> t = T()
>>> t.sig.pause()
>>> t.sig.emit(1)
>>> t.sig.emit(2)
>>> t.sig.emit(3)
>>> t.sig.resume(lambda a, b: (a[0].union(set(b)),), (set(),))
>>> # results in t.sig.emit({1, 2, 3})
Source code in psygnal/_signal.py
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
def resume(self, reducer: ReducerFunc | None = None, initial: Any = _NULL) -> None:
    """Resume (unpause) this signal, emitting everything in the queue.

    Parameters
    ----------
    reducer : Callable | None
        A optional function to reduce the args collected while paused into a single
        emitted group of args.  If not provided, all emissions will be re-emitted
        as they were collected when the signal is resumed. May be:

        - a function that takes two args tuples and returns a single args tuple.
          This will be passed to `functools.reduce` and is expected to reduce all
          collected/emitted args into a single tuple.
          For example, three `emit(1)` events would be reduced and re-emitted as
          follows: `self.emit(*functools.reduce(reducer, [(1,), (1,), (1,)]))`
        - a function that takes a single argument (an iterable of args tuples) and
          returns a tuple (the reduced args). This will be *not* be passed to
          `functools.reduce`. If `reducer` is a function that takes a single
          argument, `initial` will be ignored.
    initial: any, optional
        initial value to pass to `functools.reduce`

    Examples
    --------
    >>> class T:
    ...     sig = Signal(int)
    >>> t = T()
    >>> t.sig.pause()
    >>> t.sig.emit(1)
    >>> t.sig.emit(2)
    >>> t.sig.emit(3)
    >>> t.sig.resume(lambda a, b: (a[0].union(set(b)),), (set(),))
    >>> # results in t.sig.emit({1, 2, 3})
    """
    self._is_paused = False
    # not sure why this attribute wouldn't be set, but when resuming in
    # EventedModel.update, it may be undefined (as seen in tests)
    if not getattr(self, "_args_queue", None):
        return
    if len(self._slots) == 0:
        self._args_queue.clear()
        return

    if reducer is not None:
        if len(inspect.signature(reducer).parameters) == 1:
            args = cast("ReducerOneArg", reducer)(self._args_queue)
        else:
            reducer = cast("ReducerTwoArgs", reducer)
            if initial is _NULL:
                args = reduce(reducer, self._args_queue)
            else:
                args = reduce(reducer, self._args_queue, initial)
        self._run_emit_loop(args)
    else:
        for args in self._args_queue:
            self._run_emit_loop(args)
    self._args_queue.clear()

unblock() #

Unblock this signal, allowing it to emit.

Source code in psygnal/_signal.py
969
970
971
def unblock(self) -> None:
    """Unblock this signal, allowing it to emit."""
    self._is_blocked = False

psygnal.SignalGroup #

A collection of signals that can be connected to as a single unit.

This class is not intended to be instantiated directly. Instead, it should be subclassed, and the subclass should define Signal instances as class attributes. The SignalGroup will then automatically collect these signals and provide a SignalRelay instance (at group.all) that can be used to connect to all of the signals in the group.

This class is used in both the EventedModels and the evented dataclass patterns. See also: psygnal.SignalGroupDescriptor, which provides convenient and explicit way to create a SignalGroup on a dataclass-like class.

Parameters:

  • instance (Any, optional) –

    An object to which this SignalGroup is bound, by default None

Attributes:

  • all (SignalRelay) –

    A special SignalRelay instance that can be used to connect to all signals in this group.

Examples:

from psygnal import Signal, SignalGroup

class MySignals(SignalGroup):
    sig1 = Signal()
    sig2 = Signal()

group = MySignals()
group.all.connect(print) # connect to all signals in the group

list(group)                  # ['sig1', 'sig2']
len(group)                   # 2
group.sig1 is group['sig1']  # True
Source code in psygnal/_group.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
@mypyc_attr(allow_interpreted_subclasses=True)
class SignalGroup:
    """A collection of signals that can be connected to as a single unit.

    This class is not intended to be instantiated directly.  Instead, it should be
    subclassed, and the subclass should define Signal instances as class attributes.
    The SignalGroup will then automatically collect these signals and provide a
    SignalRelay instance (at `group.all`) that can be used to connect to all of the
    signals in the group.

    This class is used in both the EventedModels and the evented dataclass patterns.
    See also: `psygnal.SignalGroupDescriptor`, which provides convenient and explicit
    way to create a SignalGroup on a dataclass-like class.

    Parameters
    ----------
    instance : Any, optional
        An object to which this `SignalGroup` is bound, by default None

    Attributes
    ----------
    all : SignalRelay
        A special SignalRelay instance that can be used to connect to all signals in
        this group.

    Examples
    --------
    ```python
    from psygnal import Signal, SignalGroup

    class MySignals(SignalGroup):
        sig1 = Signal()
        sig2 = Signal()

    group = MySignals()
    group.all.connect(print) # connect to all signals in the group

    list(group)                  # ['sig1', 'sig2']
    len(group)                   # 2
    group.sig1 is group['sig1']  # True
    ```
    """

    _psygnal_signals: ClassVar[Mapping[str, Signal]]
    _psygnal_instances: dict[str, SignalInstance]
    _psygnal_uniform: ClassVar[bool] = False

    def __init__(self, instance: Any = None) -> None:
        cls = type(self)
        if not hasattr(cls, "_psygnal_signals"):  # pragma: no cover
            raise TypeError(
                "Cannot instantiate `SignalGroup` directly.  Use a subclass instead."
            )

        self._psygnal_instances = {
            name: sig.__get__(self, cls) for name, sig in cls._psygnal_signals.items()
        }
        self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)

    def __init_subclass__(cls, strict: bool = False) -> None:
        """Collects all Signal instances on the class under `cls._psygnal_signals`."""
        cls._psygnal_signals = {
            k: val
            for k, val in getattr(cls, "__dict__", {}).items()
            if isinstance(val, Signal)
        }

        if conflicts := {k for k in cls._psygnal_signals if k.startswith("_psygnal")}:
            warnings.warn(
                "Signal names may not begin with '_psygnal'. "
                f"Skipping signals: {conflicts}",
                stacklevel=2,
            )
            for key in conflicts:
                del cls._psygnal_signals[key]

        if "all" in cls._psygnal_signals:
            warnings.warn(
                "Name 'all' is reserved for the SignalRelay. You cannot use this "
                "name on to access a SignalInstance on a SignalGroup. (You may still "
                "access it at `group['all']`).",
                UserWarning,
                stacklevel=2,
            )
            delattr(cls, "all")

        if "psygnals_uniform" in cls._psygnal_signals:
            raise NameError(
                "Name 'psygnals_uniform' is reserved.  You cannot use this "
                "name as a signal on a SignalGroup"
            )

        cls._psygnal_uniform = _is_uniform(cls._psygnal_signals.values())
        if strict and not cls._psygnal_uniform:
            raise TypeError(
                "All Signals in a strict SignalGroup must have the same signature"
            )
        super().__init_subclass__()

    @property
    def all(self) -> SignalRelay:
        """SignalInstance that can be used to connect to all signals in this group.

        Examples
        --------
        ```python
        from psygnal import Signal, SignalGroup

        class MySignals(SignalGroup):
            sig1 = Signal()
            sig2 = Signal()

        group = MySignals()
        group.sig2.connect(...)  # connect to a single signal by name
        group.all.connect(...)  # connect to all signals in the group
        """
        return self._psygnal_relay

    # TODO: change type hint to -> SignalInstance after completing deprecation of
    # direct access to names on SignalRelay object
    def __getattr__(self, name: str) -> Any:
        # Note, technically these lines aren't actually needed because of Signal's
        # descriptor protocol: Accessing a name on a group instance will first look
        # the instance's __dict__, and then in the class's __dict__, which
        # will call Signal.__get__ and return the SignalInstance.
        # these lines are here as a reminder to developers (and safeguard?).
        if name != "_psygnal_instances" and name in self._psygnal_instances:
            return self._psygnal_instances[name]  # pragma: no cover

        if name != "_psygnal_relay" and hasattr(self._psygnal_relay, name):
            warnings.warn(
                f"Accessing SignalInstance attribute {name!r} on a SignalGroup is "
                f"deprecated. Access it on the `group.all` attribute instead. e.g. "
                f"`group.all.{name}`. This will be an error in v0.11.",
                FutureWarning,
                stacklevel=2,
            )
            return getattr(self._psygnal_relay, name)

        raise AttributeError(f"{type(self).__name__!r} has no signal named {name!r}")

    @property
    def signals(self) -> Mapping[str, SignalInstance]:
        """DEPRECATED: A mapping of signal names to SignalInstance instances."""
        # TODO: deprecate this property
        warnings.warn(
            "Accessing the `signals` property on a SignalGroup is deprecated. "
            "Use __iter__ to iterate over all signal names, and __getitem__ or getattr "
            "to access signal instances. This will be an error in a future.",
            FutureWarning,
            stacklevel=2,
        )
        return self._psygnal_instances

    def __len__(self) -> int:
        """Return the number of signals in the group (not including the relay)."""
        return len(self._psygnal_signals)

    def __getitem__(self, item: str) -> SignalInstance:
        """Get a signal instance by name."""
        return self._psygnal_instances[item]

    def __iter__(self) -> Iterator[str]:
        """Yield the names of all signals in the group."""
        return iter(self._psygnal_signals)

    def __contains__(self, item: str) -> bool:
        """Return True if the group contains a signal with the given name."""
        # this is redundant with __iter__ and can be removed, but only after
        # removing the deprecation warning in __getattr__
        return item in self._psygnal_signals

    def __repr__(self) -> str:
        """Return repr(self)."""
        name = self.__class__.__name__
        return f"<SignalGroup {name!r} with {len(self)} signals>"

    @classmethod
    def psygnals_uniform(cls) -> bool:
        """Return true if all signals in the group have the same signature."""
        return cls._psygnal_uniform

    @classmethod
    def is_uniform(cls) -> bool:
        """Return true if all signals in the group have the same signature."""
        warnings.warn(
            "The `is_uniform` method on SignalGroup is deprecated. Use "
            "`psygnals_uniform` instead. This will be an error in v0.11.",
            FutureWarning,
            stacklevel=2,
        )
        return cls._psygnal_uniform

    def __deepcopy__(self, memo: dict[int, Any]) -> SignalGroup:
        # TODO:
        # This really isn't a deep copy. Should we also copy connections?
        # a working deepcopy is important for pydantic support, but in most cases
        # it will be a group without any signals connected
        return type(self)(instance=self._psygnal_relay.instance)

all: SignalRelay property #

SignalInstance that can be used to connect to all signals in this group.

Examples:

```python from psygnal import Signal, SignalGroup

class MySignals(SignalGroup): sig1 = Signal() sig2 = Signal()

group = MySignals() group.sig2.connect(...) # connect to a single signal by name group.all.connect(...) # connect to all signals in the group

signals: Mapping[str, SignalInstance] property #

DEPRECATED: A mapping of signal names to SignalInstance instances.

is_uniform() classmethod #

Return true if all signals in the group have the same signature.

Source code in psygnal/_group.py
392
393
394
395
396
397
398
399
400
401
@classmethod
def is_uniform(cls) -> bool:
    """Return true if all signals in the group have the same signature."""
    warnings.warn(
        "The `is_uniform` method on SignalGroup is deprecated. Use "
        "`psygnals_uniform` instead. This will be an error in v0.11.",
        FutureWarning,
        stacklevel=2,
    )
    return cls._psygnal_uniform

psygnals_uniform() classmethod #

Return true if all signals in the group have the same signature.

Source code in psygnal/_group.py
387
388
389
390
@classmethod
def psygnals_uniform(cls) -> bool:
    """Return true if all signals in the group have the same signature."""
    return cls._psygnal_uniform

psygnal.EmissionInfo #

Bases: NamedTuple

Tuple containing information about an emission event.

Attributes:

Source code in psygnal/_group.py
37
38
39
40
41
42
43
44
45
46
47
class EmissionInfo(NamedTuple):
    """Tuple containing information about an emission event.

    Attributes
    ----------
    signal : SignalInstance
    args: tuple
    """

    signal: SignalInstance
    args: tuple[Any, ...]

psygnal.EmitLoopError #

Bases: Exception

Error type raised when an exception occurs during a callback.

Source code in psygnal/_exceptions.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class EmitLoopError(Exception):
    """Error type raised when an exception occurs during a callback."""

    def __init__(
        self,
        cb: WeakCallback | Callable,
        args: tuple,
        exc: BaseException,
        signal: SignalInstance | None = None,
    ) -> None:
        self.exc = exc
        self.args = args
        self.__cause__ = exc  # mypyc doesn't set this, but uncompiled code would
        if signal is None:
            sig_name = ""
        else:
            inst_class = signal.instance.__class__
            mod = getattr(inst_class, "__module__", "")
            sig_name = f"{mod}.{inst_class.__qualname__}.{signal.name}"
        if isinstance(cb, WeakCallback):
            cb_name = cb.slot_repr()
        else:
            cb_name = getattr(cb, "__qualname__", repr(cb))
        super().__init__(
            MSG.format(
                sig=sig_name,
                cb=cb_name,
                args=args,
                err=exc.__class__.__name__,
            )
        )