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!!