<commit_before>""" Event management system.

This module provides API for event management. There are two APIs provided:

* Global event management API: subscribe, unsubscribe, handle.
* Local event management API: Manager

If you run only one instance of your application per Python
interpreter you can stick with global API, but if you want to have
more than one application instances running inside one interpreter and
to have different configurations for them -- you should use local API
and have one instance of Manager object per application instance.
"""

from typing import Callable, Set, Type

from gaphor.misc.generic.registry import Registry, TypeAxis


__all__ = "Manager"

Event = object
Handler = Callable[[object], None]
HandlerSet = Set[Handler]


class Manager:
    """ Event manager

    Provides API for subscribing for and firing events. There's also global
    event manager instantiated at module level with functions
    :func:`.subscribe`, :func:`.handle` and decorator :func:`.subscriber` aliased
    to corresponding methods of class.
    """

    def __init__(self) -> None:
        axes = (("event_type", TypeAxis()),)
        self.registry = Registry[HandlerSet](*axes)

    def subscribe(self, handler: Handler, event_type: Type[Event]) -> None:
        """ Subscribe ``handler`` to specified ``event_type``"""
        handler_set = self.registry.get_registration(event_type)
        if handler_set is None:
            handler_set = self._register_handler_set(event_type)
        handler_set.add(handler)

    def unsubscribe(self, handler: Handler, event_type: Type[Event]) -> None:
        """ Unsubscribe ``handler`` from ``event_type``"""
        handler_set = self.registry.get_registration(event_type)
        if handler_set and handler in handler_set:
            handler_set.remove(handler)

    def handle(self, event: Event) -> None:
        """ Fire ``event``

        All subscribers will be executed with no determined order.
        """
        handler_sets = self.registry.query(event)
        for handler_set in handler_sets:
            if handler_set:
                for handler in set(handler_set):
                    handler(event)

    def _register_handler_set(self, event_type: Type[Event]) -> HandlerSet:
        """ Register new handler set for ``event_type``.
        """
        handler_set: HandlerSet = set()<commit_msg>Fix type hint for registry in event manager<commit_after>""" Event management system.

This module provides API for event management. There are two APIs provided:

* Global event management API: subscribe, unsubscribe, handle.
* Local event management API: Manager

If you run only one instance of your application per Python
interpreter you can stick with global API, but if you want to have
more than one application instances running inside one interpreter and
to have different configurations for them -- you should use local API
and have one instance of Manager object per application instance.
"""

from typing import Callable, Set, Type

from gaphor.misc.generic.registry import Registry, TypeAxis


__all__ = "Manager"

Event = object
Handler = Callable[[object], None]
HandlerSet = Set[Handler]


class Manager:
    """ Event manager

    Provides API for subscribing for and firing events. There's also global
    event manager instantiated at module level with functions
    :func:`.subscribe`, :func:`.handle` and decorator :func:`.subscriber` aliased
    to corresponding methods of class.
    """

    registry: Registry[HandlerSet]

    def __init__(self) -> None:
        axes = (("event_type", TypeAxis()),)
        self.registry = Registry(*axes)

    def subscribe(self, handler: Handler, event_type: Type[Event]) -> None:
        """ Subscribe ``handler`` to specified ``event_type``"""
        handler_set = self.registry.get_registration(event_type)
        if handler_set is None:
            handler_set = self._register_handler_set(event_type)
        handler_set.add(handler)

    def unsubscribe(self, handler: Handler, event_type: Type[Event]) -> None:
        """ Unsubscribe ``handler`` from ``event_type``"""
        handler_set = self.registry.get_registration(event_type)
        if handler_set and handler in handler_set:
            handler_set.remove(handler)

    def handle(self, event: Event) -> None:
        """ Fire ``event``

        All subscribers will be executed with no determined order.
        """
        handler_sets = self.registry.query(event)
        for handler_set in handler_sets:
            if handler_set:
                for handler in set(handler_set):
                    handler(event)

    def _register_handler_set(self, event_type: Type[Event]) -> HandlerSet:
        """ Register new handler set for ``event_type``.
        """
        handler_set: HandlerSet = set()
{'fraction_non_alphanumeric': 0.09028642590286426, 'fraction_numerical': 0.0, 'mean_word_length': 3.2759538598047917, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1754738', 'n_tokens_mistral': 1279, 'n_tokens_neox': 1216, 'n_words': 466}