How to use _get_event_name method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

mediator.py

Source:mediator.py

...15            for listener in self._listeners[event_name].values():16                listener(event)17        return event18    def add_listener(self, event, listener, priority=None):19        event_name = _get_event_name(event)20        if event_name not in self._listeners:21            self._listeners[event_name] = {}22        if priority is not None:23            if priority in self._listeners[event_name]:24                raise IndexError(25                    'The event "{0}" already have a listener '26                    'with priority {1}'.format(27                        event_name,28                        priority29                    )30                )31        else:32            if len(self._listeners[event_name]):33                priority = list(self._listeners[event_name].keys())[-1] + 134            else:35                priority = 036        self._listeners[event_name][priority] = listener37        self._listeners[event_name] = collections.OrderedDict(sorted(38            self._listeners[event_name].items(),39            key=lambda item: item[0]40        ))41    def remove_listener(self, event, listener=None):42        event_name = _get_event_name(event)43        if event_name in self._listeners:44            if not listener:45                del self._listeners[event_name]46            else:47                for p, l in self._listeners[event_name].items():48                    if l is listener:49                        self._listeners[event_name].pop(p)50                        return51    def add_subscriber(self, subscriber):52        if not isinstance(subscriber, SubscriberInterface):53            raise TypeError('Expects instance of SubscriberInterface')54        for event_name, params in subscriber.get_subscribed_events().items():55            if isinstance(params, str):56                self.add_listener(event_name, getattr(subscriber, params))57            elif isinstance(params, list):58                if not params:59                    raise ValueError('Invalid params "%r" for event "%s"' % (params, event_name))60                if len(params) <= 2 and isinstance(params[0], str):61                    priority = params[1] if len(params) > 1 else None62                    self.add_listener(event_name, getattr(subscriber, params[0]), priority)63                else:64                    for listener in params:65                        priority = listener[1] if len(listener) > 1 else None66                        self.add_listener(event_name, getattr(subscriber, listener[0]), priority)67            else:68                raise ValueError('Invalid params for event "%s"' % event_name)69class Event(object):70    event_name = None71    @classmethod72    def get_event_name(cls):73        return cls.__name__ if cls.event_name is None else cls.event_name74    @classmethod75    def listen(cls, priority=None, instance='mediator', category=VENUSIAN_CATEGORY):76        event_name = cls.get_event_name()77        def decorator(listener):78            def callback(scanner, name, ob):79                getattr(scanner, instance).add_listener(event_name, listener, priority)80            venusian.attach(listener, callback, category=category)81            return listener82        return decorator83class SubscriberInterface(object):84    def get_subscribed_events(self):85        raise NotImplementedError()86def _get_event_name(event):87    if isinstance(event, str):88        return event89    if inspect.isclass(event) and issubclass(event, Event):90        return event.get_event_name()...

events.py

Source:events.py

...70        return self._event_registry.on(event, callback)71    def push(self, payload: Payload):72        self._stream.push(payload)73    def _create_event(self, payload: Payload) -> events.EventModel:74        name = self._get_event_name(payload.event)75        model = getattr(events, name, None)76        if not model:77            print("COULDN'T FIND", name)78            return events.make_event_object(name, payload.event)79        try:80            return model(**payload.data)81        except:82            import pprint83            print(payload.event)84            pprint.pp(payload.data)85            raise86    def _event_filter_factory(self, *event_names: str) -> Filter:87        return (88            lambda payload: event_names[0].casefold() in {"any", "all", "*"}89            or payload.event in event_names90        )91    def _get_event_name(self, event: str) -> str:92        cleaned = "".join(section.capitalize() for section in event.split("_"))93        return f"Event{cleaned}"94    def _only_discord_events(self, payload: Payload) -> bool:95        return bool(payload.event)96    def _raw_filter_factory(self, *, op_code: int = None, event: str = None) -> Filter:97        return lambda payload: (payload.op_code == op_code or op_code is None) and (98            payload.event == event or event is None...

FluffyEventSystem.py

Source:FluffyEventSystem.py

...4class EventHook(object):5    def __init__(self):6        self.events = defaultdict(list)7    @staticmethod8    def _get_event_name(event: str or BaseEvent):9        """ A helper function for determining events name"""10        if type(event) == str:11            return event12        return event.event_name13    def append_event(self, event: str or BaseEvent):14        event_name = self._get_event_name(event)15        if event_name not in self.events:16            self.events[self._get_event_name(event_name)] = []17    def append_listener(self, event: str or BaseEvent, listener):18        event_name = self._get_event_name(event)19        if listener not in self.events[event_name]:20            self.append_event(event_name)21        if event_name in self.events.keys() and listener not in self.events[event_name]:22            self.events[event_name].append(listener)23    def remove_event(self, event: str or BaseEvent):24        event_name = self._get_event_name(event)25        if event_name in self.events:26            self.events.pop(event_name)27    def remove_listener(self, event: str or BaseEvent, listener):28        event_name = self._get_event_name(event)29        if event_name in self.events:30            if listener in self.events[event_name]:31                self.events[event_name].remove(listener)32    async def notify(self, event: str or BaseEvent, *args, **kwargs):33        event_name = self._get_event_name(event)34        if not len(self.events[event_name]):35            pass36        if self.events[event_name]:37            for listener in self.events[event_name]:38                if inspect.iscoroutinefunction(listener):39                    await listener(*args, **kwargs)40                else:41                    listener(*args, **kwargs)42def setup():...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.