Discover AI-native insights at Testμ Conference‘25

How to use _add_event_handler method in Playwright Python

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()...

StackOverFlow community discussions

Questions

Discussion

How to select an input according to a parent sibling label

Playwright with python - Download file from CloudFlare

Python playwright won't let me open user profile other than default profile

How to check for element existence without getting an error in Playwright

Handling pagination in python playwright when the url doesn't change

how to get element from a table, to be selected in playwright python

bytesio type abnormal Two pictures saved in the same way are of different types ： <class '_io.BytesIO'> and <class 'bytes'>

Website Access Denied in My RPA Python/Playwright

How to scrape dynamic content from a website?

Python Playwright API get hrefs

If you try to perform an action on a label which has associated input element Playwright will automatically retarget the action to the input.

If the DOM structure is always the same you can try something like this

page.locator('div:has( > :text-matches("Project")) + div input')

Otherwise layout selector is your best bet.

https://stackoverflow.com/questions/70986996/how-to-select-an-input-according-to-a-parent-sibling-label

Not guaranteed for success. Change the parameters in function async_cf_retry if it tends to fail. I wrote this code by modifying my heavy modules, so it may not work elegantly.

import re
import asyncio
from playwright.async_api import async_playwright, Error, Page
from cf_clearance import stealth_async
import httpx

# local_client = httpx.AsyncClient(verify=False)


async def async_cf_retry(page: Page, tries=10) -> bool:
    # use tries=-1 for infinite retries
    # excerpted from `from cf_clearance import async_retry`
    success = False
    while tries != 0:
        try:
            title = await page.title()
        except Error:
            tries -= 1
            await asyncio.sleep(1)
        else:
            # print(title)
            if title == 'Please Wait... | Cloudflare':
                await page.close()
                raise NotImplementedError('Encountered recaptcha. Check whether your proxy is an elite proxy.')
            elif title == 'Just a moment...':
                tries -= 1
                await asyncio.sleep(5)
            elif "www." in title:
                await page.reload()
                tries -= 1
                await asyncio.sleep(5)
            else:
                success = True
                break
    return success


async def get_client_with_clearance(proxy: str = None):
    async def get_one_clearance(proxy=proxy, logs=False):
        # proxy = {"server": "socks5://localhost:7890"}
        if type(proxy) is str:
            proxy = {'server': proxy}
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False, proxy=proxy,
                                              args=["--window-position=1000,800", "--disable-web-security",
                                                    "--disable-webgl"])
            page = await browser.new_page()#viewport={"width": 0, "height": 0})
            await stealth_async(page)
            
            # Download
            async def on_download(download):
                print('download', download)
                await download.save_as(download.suggested_filename)
            page.on("download", on_download)

            if logs:
                def log_response(intercepted_response):
                    print("a response was received:", intercepted_response.url)
                
                page.on("response", log_response)
            
            await page.goto("https://www.historique-meteo.net/site/export.php?ville_id=1067")
            res = await async_cf_retry(page)
            if res:
                cookies = await page.context.cookies()
                cookies_for_httpx = {cookie['name']: cookie['value'] for cookie in cookies}
                ua = await page.evaluate('() => {return navigator.userAgent}')
                # print(ua)
            else:
                await page.close()
                raise InterruptedError("cf challenge fail")
            await asyncio.sleep(10000)
            return ua, cookies_for_httpx
    
    ua, cookies_for_httpx = await get_one_clearance(logs=True)
    print(cookies_for_httpx)


print(asyncio.get_event_loop().run_until_complete(get_client_with_clearance(
    # proxy='http://localhost:8888'
    # use proxifier on windows as an elite proxy
)))

If download <Download url='https://www.historique-meteo.net/site/export.php?ville_id=1067' suggested_filename='export-aix-les-bains.csv'> is shown in the console, the browser must be downloading your file.
This code may keep trying to get a cloudflare clearance even if the file has been downloaded. Edit your strategies in async_cf_retry if this happens.
Use persisent context offered by playwright to save and reuse your cloudflare clearance. Consider downloading with httpx.

https://stackoverflow.com/questions/70734016/playwright-with-python-download-file-from-cloudflare

You could use something called element_handle.is_visible() from here.

your code should look something like below.

href_element = page.query_selector("a")
href_element.is_visible()

https://stackoverflow.com/questions/73181978/how-to-check-for-element-existence-without-getting-an-error-in-playwright

With some small adjustments you can get it, lets try this:

from bs4 import BeautifulSoup as bs
from playwright.sync_api import sync_playwright
import time

url = 'https://franchisedisclosure.gov.au/Register'

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False, slow_mo=100)
    page = browser.new_page()
    page.goto(url)
    page.locator("text=I agree to the terms of use").click()
    page.locator("text=Continue").click()
    page.wait_for_load_state('domcontentloaded')
    names = []
    industry = []
    Locations = []
    # When you click to next page, an element with text "Loading" appears in the screen, so we save that element
    loading_icon = "//strong[text()='Loading...']"
    # This is the "next page" button
    next_page_locator = "//ul[@class='pagination']/li[3]"
    # We select the option of 50 elements per page
    page.select_option("#perPageCount", value="50")
    # We wait for the selector of loading icon to be visible and then to be hidden, which means the new list is fully loaded
    page.wait_for_selector(loading_icon, state="visible")
    page.wait_for_selector(loading_icon, state="hidden")
    time.sleep(1)
    # We make a loop until the button "Next page" is disabled, which means there are no more pages to paginate
    while "disabled" not in page.get_attribute(selector=next_page_locator, name="class"):
        # We get the info you wanted
        page.is_visible('tbody')
        html = page.inner_html('table.table.table-hover')
        soup = bs(html, 'html.parser')
        table = soup.find('tbody')
        rows = table.findAll('tr')
        for row in rows:
            info = row.findAll('td')
            name = info[0].text.strip()
            industry = info[1].text.strip()
            Locations = info[2].text.strip()
        # Once we get the info we click in next page and we wait for the loading element to be visible and then to be hidden.
        page.click(next_page_locator)
        page.wait_for_selector(loading_icon, state="visible")
        page.wait_for_selector(loading_icon, state="hidden")
        time.sleep(1)

https://stackoverflow.com/questions/74696134/handling-pagination-in-python-playwright-when-the-url-doesnt-change

To get the id you will need the parent, and evaluate the id attribute

const td = page.locator("text=Edição 993 link direto gdb");
const tr = await td.$('xpath=..')
const id = await tr.evaluate(node => node.getAttribute('id'))

https://stackoverflow.com/questions/71651784/how-to-get-element-from-a-table-to-be-selected-in-playwright-python

this worked for me in java:

Page page = browser.newPage();
Map<String, String> headers = new HashMap<>();
headers.put("user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) `enter code here`AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36");
page.setExtraHTTPHeaders(headers );

https://stackoverflow.com/questions/72088427/website-access-denied-in-my-rpa-python-playwright

You can do:

hrefs_of_page = page.eval_on_selector_all("a[href^='/wiki/']", "elements => elements.map(element => element.href)")

which should work for your use-case. This will lookup for all the link tags which have a href attribute which starts with /wiki. Then on the browser side JavaScript gets evaluated which maps from an array of elements to the href attribute so a string array gets returned on the Python side.

https://stackoverflow.com/questions/68013586/python-playwright-api-get-hrefs

Blogs

Check out the latest blogs from LambdaTest on this topic:

Getting Started With Automation Testing Using Selenium Ruby

Ruby is a programming language which is well suitable for web automation. Ruby makes an excellent choice because of its clean syntax, focus on built-in library integrations, and an active community. Another benefit of Ruby is that it also allows other programming languages like Java, Python, etc. to be used in order to automate applications written in any other frameworks. Therefore you can use Selenium Ruby to automate any sort of application in your system and test the results in any type of testing environment

Playwright End To End Testing Tutorial: A Complete Guide

It is essential for a team, when speaking about test automation, to take the time needed to think, analyze and try what will be the best tool, framework, and language that suits your team’s needs.

How To Choose The Best JavaScript Unit Testing Frameworks

JavaScript is one of the most widely used programming languages. This popularity invites a lot of JavaScript development and testing frameworks to ease the process of working with it. As a result, numerous JavaScript testing frameworks can be used to perform unit testing.

11 Best Automated UI Testing Tools In 2022

The web development industry is growing, and many Best Automated UI Testing Tools are available to test your web-based project to ensure it is bug-free and easily accessible for every user. These tools help you test your web project and make it fully compatible with user-end requirements and needs.

Selenium 4.0 and The Future: Manoj Kumar [Testμ 2022]

We were eager to listen to Manoj Kumar, VP Developer Relations, LambdaTest, speak on the importance of Selenium 4.0 and how bright the future is. This was the agenda of the speech:

Playwright tutorial

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.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.