Best Python code snippet using playwright-python
event_handlers.py
Source:event_handlers.py  
...16    """Responsible for invoking event handlers for given events."""17    def __init__(self):18        self._event_handler_map = defaultdict(list)19        self._build()20    def _add_event_handler(self, handler):21        self._event_handler_map[handler.event_id].append(handler)22    def _build(self):23        self._add_event_handler(LoginEventHandler())24        self._add_event_handler(TurnOffNotificationEventHandler())25        event_to_message_mapping = database.get_event_to_message_mapping()26        for event_id, message_ids in event_to_message_mapping.iteritems():27            self._add_event_handler(EventToMessagesHandler(event_id, message_ids))28    def get_handlers(self, event_id):29        return self._event_handler_map[event_id]30    def get_event_ids(self):31        return self._event_handler_map.keys()32class EventHandler():33    def __init__(self, event_id):34        self.event_id = event_id35    def handle_event(self, event, event_params):36        raise Exception("Not implemented!")37    def validate(self, event, event_params):38        return event.has_field('user_id') and event.has_field('timestamp')39class LoginEventHandler(EventHandler):40    """Writes user data to database on login event."""41    def __init__(self):...base.py
Source:base.py  
...62        By only supporting the decorator use case, this method has improved63        type safety over ``EventEmitter#on``.64        """65        def on(f: Handler) -> Handler:66            self._add_event_handler(event, f, f)67            return f68        return on69    def add_listener(self, event: str, f: Handler) -> Handler:70        """Register the function ``f`` to the event name ``event``::71            def data_handler(data):72                print(data)73            h = ee.add_listener("event", data_handler)74        By not supporting the decorator use case, this method has improved75        type safety over ``EventEmitter#on``.76        """77        self._add_event_handler(event, f, f)78        return f79    def _add_event_handler(self, event: str, k: Callable, v: Callable):80        # Fire 'new_listener' *before* adding the new listener!81        self.emit("new_listener", event, k)82        # Add the necessary function83        # Note that k and v are the same for `on` handlers, but84        # different for `once` handlers, where v is a wrapped version85        # of k which removes itself before calling k86        with self._lock:87            if event not in self._events:88                self._events[event] = OrderedDict()89            self._events[event][k] = v90    def _emit_run(91        self,92        f: Callable,93        args: Tuple[Any, ...],94        kwargs: Dict[str, Any],95    ) -> None:96        f(*args, **kwargs)97    def event_names(self) -> Set[str]:98        """Get a set of events that this emitter is listening to."""99        return set(self._events.keys())100    def _emit_handle_potential_error(self, event: str, error: Any) -> None:101        if event == "error":102            if isinstance(error, Exception):103                raise error104            else:105                raise PyeeException(f"Uncaught, unspecified 'error' event: {error}")106    def _call_handlers(107        self,108        event: str,109        args: Tuple[Any, ...],110        kwargs: Dict[str, Any],111    ) -> bool:112        handled = False113        with self._lock:114            funcs = list(self._events.get(event, OrderedDict()).values())115        for f in funcs:116            self._emit_run(f, args, kwargs)117            handled = True118        return handled119    def emit(120        self,121        event: str,122        *args: Any,123        **kwargs: Any,124    ) -> bool:125        """Emit ``event``, passing ``*args`` and ``**kwargs`` to each attached126        function. Returns ``True`` if any functions are attached to ``event``;127        otherwise returns ``False``.128        Example::129            ee.emit('data', '00101001')130        Assuming ``data`` is an attached function, this will call131        ``data('00101001')'``.132        """133        handled = self._call_handlers(event, args, kwargs)134        if not handled:135            self._emit_handle_potential_error(event, args[0] if args else None)136        return handled137    def once(138        self,139        event: str,140        f: Callable = None,141    ) -> Callable:142        """The same as ``ee.on``, except that the listener is automatically143        removed after being called.144        """145        def _wrapper(f: Callable) -> Callable:146            def g(147                *args: Any,148                **kwargs: Any,149            ) -> Any:150                with self._lock:151                    # Check that the event wasn't removed already right152                    # before the lock153                    if event in self._events and f in self._events[event]:154                        self._remove_listener(event, f)155                    else:156                        return None157                # f may return a coroutine, so we need to return that158                # result here so that emit can schedule it159                return f(*args, **kwargs)160            self._add_event_handler(event, f, g)161            return f162        if f is None:163            return _wrapper164        else:165            return _wrapper(f)166    def _remove_listener(self, event: str, f: Callable) -> None:167        """Naked unprotected removal."""168        self._events[event].pop(f)169        if not len(self._events[event]):170            del self._events[event]171    def remove_listener(self, event: str, f: Callable) -> None:172        """Removes the function ``f`` from ``event``."""173        with self._lock:174            self._remove_listener(event, f)...event_emitter.py
Source:event_emitter.py  
...63        directly, as well as use them in remove_listener calls.64        """65        with self._event_lock:66            def _on(f):67                self._add_event_handler(event, f, f)68                return f69            if f is None:70                return _on71            else:72                return _on(f)73    def _add_event_handler(self, event, k, v):74        # Fire 'new_listener' *before* adding the new listener!75        self.emit('new_listener', event, k)76        # Add the necessary function77        # Note that k and v are the same for `on` handlers, but78        # different for `once` handlers, where v is a wrapped version79        # of k which removes itself before calling k80        self._events[event][k] = v81    def emit(self, event, *args, **kwargs):82        """Emit ``event``, passing ``*args`` and ``**kwargs`` to each attached83        function. Returns ``True`` if any functions are attached to ``event``;84        otherwise returns ``False``.85        Example::86            ee.emit('data', '00101001')87        Assuming ``data`` is an attached function, this will call88        ``data('00101001')'``.89        For coroutine event handlers, calling emit is non-blocking. In other90        words, you do not have to await any results from emit, and the91        coroutine is scheduled in a fire-and-forget fashion.92        """93        handled = False94        with self._event_lock:95            for f in list(self._events[event].values()):96                result = f(*args, **kwargs)97                # If f was a coroutine function, we need to schedule it and98                # handle potential errors99                if iscoroutine and iscoroutine(result):100                    if self._loop:101                        d = self._schedule(result, loop=self._loop)102                    else:103                        d = self._schedule(result)104                    # scheduler gave us an asyncio Future105                    if hasattr(d, 'add_done_callback'):106                        @d.add_done_callback107                        def _callback(f):108                            exc = f.exception()109                            if exc:110                                self.emit('error', exc)111                    # scheduler gave us a twisted Deferred112                    elif hasattr(d, 'addErrback'):113                        @d.addErrback114                        def _callback(exc):115                            self.emit('error', exc)116                handled = True117        if not handled and event == 'error':118            if args:119                raise args[0]120            else:121                raise EventEmitterException(122                    "Uncaught, unspecified 'error' event.")123        return handled124    def once(self, event, f=None):125        """The same as ``ee.on``, except that the listener is automatically126        removed after being called.127        """128        with self._event_lock:129            def _wrapper(f):130                def g(*args, **kwargs):131                    self.remove_listener(event, f)132                    # f may return a coroutine, so we need to return that133                    # result here so that emit can schedule it134                    return f(*args, **kwargs)135                self._add_event_handler(event, f, g)136                return f137            if f is None:138                return _wrapper139            else:140                return _wrapper(f)141    def off(self, event, f):142        """Removes the function ``f`` from ``event``."""143        self._events[event].pop(f)144    def remove_listener(self, event, f):145        """Removes the function ``f`` from ``event``."""146        self._events[event].pop(f)147    def remove_all_listeners(self, event=None):148        """Remove all listeners attached to ``event``.149        If ``event`` is ``None``, remove all listeners on all events....__init__.py
Source:__init__.py  
...75        returned. The upshot of this is that you can call decorated handlers76        directly, as well as use them in remove_listener calls.77        """78        def _on(f):79            self._add_event_handler(event, f, f)80            return f81        if f is None:82            return _on83        else:84            return _on(f)85    def _add_event_handler(self, event, k, v):86        # Fire 'new_listener' *before* adding the new listener!87        self.emit('new_listener', event, k)88        # Add the necessary function89        # Note that k and v are the same for `on` handlers, but90        # different for `once` handlers, where v is a wrapped version91        # of k which removes itself before calling k92        self._events[event][k] = v93    def emit(self, event, *args, **kwargs):94        """Emit ``event``, passing ``*args`` and ``**kwargs`` to each attached95        function. Returns ``True`` if any functions are attached to ``event``;96        otherwise returns ``False``.97        Example::98            ee.emit('data', '00101001')99        Assuming ``data`` is an attached function, this will call100        ``data('00101001')'``.101        For coroutine event handlers, calling emit is non-blocking. In other102        words, you do not have to await any results from emit, and the103        coroutine is scheduled in a fire-and-forget fashion.104        """105        handled = False106        for f in list(self._events[event].values()):107            result = f(*args, **kwargs)108            # If f was a coroutine function, we need to schedule it and109            # handle potential errors110            if iscoroutine and iscoroutine(result):111                if self._loop:112                    d = self._schedule(result, loop=self._loop)113                else:114                    d = self._schedule(result)115                # scheduler gave us an asyncio Future116                if hasattr(d, 'add_done_callback'):117                    @d.add_done_callback118                    def _callback(f):119                        exc = f.exception()120                        if exc:121                            self.emit('error', exc)122                # scheduler gave us a twisted Deferred123                elif hasattr(d, 'addErrback'):124                    @d.addErrback125                    def _callback(exc):126                        self.emit('error', exc)127            handled = True128        if not handled and event == 'error':129            if args:130                raise args[0]131            else:132                raise PyeeException("Uncaught, unspecified 'error' event.")133        return handled134    def once(self, event, f=None):135        """The same as ``ee.on``, except that the listener is automatically136        removed after being called.137        """138        def _wrapper(f):139            def g(*args, **kwargs):140                self.remove_listener(event, f)141                # f may return a coroutine, so we need to return that142                # result here so that emit can schedule it143                return f(*args, **kwargs)144            self._add_event_handler(event, f, g)145            return f146        if f is None:147            return _wrapper148        else:149            return _wrapper(f)150    def remove_listener(self, event, f):151        """Removes the function ``f`` from ``event``."""152        self._events[event].pop(f)153    def remove_all_listeners(self, event=None):154        """Remove all listeners attached to ``event``.155        If ``event`` is ``None``, remove all listeners on all events.156        """157        if event is not None:158            self._events[event] = OrderedDict()...LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!
