mitt
Reputation: 9
How can i resolve the error "Event loop is closed! Is Playwright already stopped?"
I am working with Playwright in Python and facing an issue with the retry logic when an action fails.
The idea is that if the action fails on one attempt, the code should retry up to a maximum of 8 times. For each new attempt, the proxy should be renewed, changing the IP address used.
Here is the complete code for reference:
from src.services.LoggingService import LoggingService
from playwright.sync_api import sync_playwright, TimeoutError
from decouple import config
class UserService:
@staticmethod
def update(username, password, playerName, playerPassword):
attempt = 0
max_attempts = 8
while attempt < max_attempts:
browser = None
try:
with sync_playwright() as p:
if config("PROXY_ENABLED") == "true":
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
proxy={"server": config("PROXY_URL")}
)
LoggingService.info(f"Proxy enabled: {config('PROXY_URL')}", "UserService")
else:
LoggingService.info("Proxy disabled", "UserService")
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(config("GA_LOGIN_URL"), timeout=60000)
page.fill('input.input_sizable_low.input_type_text.input_validState_1', username)
page.fill('input.input_sizable_low.input_type_password.input_with-eye.input_validState_1', password)
page.click('button.button_sizable_default.button_colors_default')
page.wait_for_selector('input.input_sizable_default.input_type_text.input_validState_1', timeout=60000)
page.fill('input.input_sizable_default.input_type_text.input_validState_1', playerName)
page.locator(f"text={playerName}").click(timeout=30000)
page.locator(".dropdown__action").first.click(timeout=30000)
page.get_by_role("link", name="Cambiar contraseña").click(timeout=30000)
page.get_by_role("textbox", name="Nueva contraseña", exact=True).fill(playerPassword)
page.get_by_role("textbox", name="Confirmar nueva contraseña").fill(playerPassword)
page.get_by_role("button", name="Cambiar").click(timeout=30000)
page.wait_for_timeout(1000)
browser.close()
LoggingService.info(f"Usuario {playerName} actualizado con éxito luego de {attempt + 1} intentos", "UserService")
return True
except Exception as e:
LoggingService.error(f"Error en intento {attempt + 1}: {str(e)}", "UserService")
attempt += 1
if attempt == max_attempts:
LoggingService.error(f"No se pudo realizar el ingreso para {playerName} tras {max_attempts} intentos", "UserService")
return False
return False
@staticmethod
def create(username, password, playerName, playerPassword):
attempt = 0
max_attempts = 8
while attempt < max_attempts:
browser = None
try:
with sync_playwright() as p:
if config("PROXY_ENABLED") == "true":
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
proxy={"server": config("PROXY_URL")}
)
LoggingService.info(f"Proxy enabled: {config('PROXY_URL')}", "UserService")
else:
LoggingService.info("Proxy disabled", "UserService")
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(config("GA_LOGIN_URL"), timeout=60000)
page.fill('input.input_sizable_low.input_type_text.input_validState_1', username)
page.fill('input.input_sizable_low.input_type_password.input_with-eye.input_validState_1', password)
page.click('button.button_sizable_default.button_colors_default')
page.get_by_role("link", name="Nuevo jugador").click(timeout=30000)
page.get_by_role("textbox", name="Nombre de usuario").fill(playerName)
page.get_by_role("textbox", name="Contraseña", exact=True).fill(playerPassword)
page.get_by_role("textbox", name="Confirmar contraseña").fill(playerPassword)
page.get_by_role("button", name="Crear jugador").click(timeout=30000)
page.locator("#modal-root").get_by_role("button", name="Crear jugador").click(timeout=30000)
page.wait_for_timeout(500)
browser.close()
LoggingService.info(f"Usuario {playerName} creado con éxito luego de {attempt + 1} intentos", "UserService")
return True
except Exception as e:
LoggingService.info(f"Error en intento {attempt + 1}: {str(e)}", "UserService")
attempt += 1
if attempt == max_attempts:
LoggingService.info(f"No se pudo realizar el ingreso para {playerName} tras {max_attempts} intentos", "UserService")
return False
return False
@staticmethod
def cashin(username: str, password: str, playerName: str, playerAmount: str):
attempt = 0
max_attempts = 8
while attempt < max_attempts:
browser = None
try:
with sync_playwright() as p:
if config("PROXY_ENABLED") == "true":
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
proxy={"server": config("PROXY_URL")}
)
LoggingService.info(f"Proxy enabled: {config('PROXY_URL')}", "UserService")
else:
LoggingService.info("Proxy disabled", "UserService")
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(config("GA_LOGIN_URL"), timeout=60000)
page.fill('input.input_sizable_low.input_type_text.input_validState_1', username)
page.fill('input.input_sizable_low.input_type_password.input_with-eye.input_validState_1', password)
page.click('button.button_sizable_default.button_colors_default')
page.wait_for_selector('input.input_sizable_default.input_type_text.input_validState_1', timeout=60000)
page.fill('input.input_sizable_default.input_type_text.input_validState_1', playerName)
page.locator(f"text={playerName}").click(timeout=30000)
page.locator("div:has-text('Depositar')").first.click(timeout=30000)
page.get_by_role("link", name="Depositar").first.click(timeout=30000)
page.get_by_role("textbox", name="Monto").wait_for(timeout=30000)
page.get_by_role("textbox", name="Monto").fill(str(playerAmount))
page.get_by_role("button", name="Depósito").click(timeout=30000)
page.wait_for_timeout(1000)
browser.close()
LoggingService.info(f"Ingreso de {playerAmount} en la cuenta @{playerName} luego de {attempt + 1} intentos", "UserService")
return True
except Exception as e:
LoggingService.error(f"Error en intento {attempt + 1}: {str(e)}", "UserService")
attempt += 1
if attempt == max_attempts:
LoggingService.error(f"No se pudo realizar el ingreso para {playerName} tras {max_attempts} intentos", "UserService")
return False
return False
@staticmethod
def cashout(username, password, playerName, playerAmount):
attempt = 0
max_attempts = 8
while attempt < max_attempts:
browser = None
try:
with sync_playwright() as p:
if config("PROXY_ENABLED") == "true":
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
proxy={"server": config("PROXY_URL")}
)
LoggingService.info(f"Proxy enabled: {config('PROXY_URL')}", "UserService")
else:
LoggingService.info("Proxy disabled", "UserService")
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(config("GA_LOGIN_URL"), timeout=60000)
page.fill('input.input_sizable_low.input_type_text.input_validState_1', username)
page.fill('input.input_sizable_low.input_type_password.input_with-eye.input_validState_1', password)
page.click('button.button_sizable_default.button_colors_default')
page.wait_for_selector('input.input_sizable_default.input_type_text.input_validState_1', timeout=60000)
page.fill('input.input_sizable_default.input_type_text.input_validState_1', playerName)
page.locator(f"text={playerName}").click(timeout=30000)
page.get_by_role("link", name="Retiro", exact=True).first.click(timeout=30000)
page.get_by_role("textbox", name="Monto").fill(str(playerAmount))
page.get_by_role("button", name="Retiro").click(timeout=30000)
page.wait_for_timeout(1000)
browser.close()
LoggingService.info(f"Retiro de {playerAmount} en la cuenta @{playerName} luego de {attempt + 1} intentos", "UserService")
return True
except Exception as e:
LoggingService.error(f"Error en intento {attempt + 1}: {str(e)}", "UserService")
attempt += 1
if attempt == max_attempts:
LoggingService.error(f"No se pudo realizar el ingreso para {playerName} tras {max_attempts} intentos", "UserService")
return False
return False
However, the code is not working as expected. I am encountering the following error:
line 104, in _sync
raise Error("Event loop is closed! Is Playwright already stopped?")
playwright._impl._errors.Error: Event loop is closed! Is Playwright already stopped?
I don’t understand why this error occurs since I am using sync_playwright() inside each attempt and properly closing the browser at the end of the process.
The main issue is that although I expect Playwright to start and close correctly on each attempt, it seems that at some point, the event loop is already closed and cannot be restarted. This prevents the reconnection attempts from working as intended.
Why is the event loop closing unexpectedly? How do I handle retries in Playwright to ensure that the proxy is properly renewed on each attempt without causing this error?
Upvotes: -1
Views: 46
Answers (2)
Akhil Chandran
Reputation: 200
The problem is that sync_playwright() should not be instantiated multiple times within a loop. Each time sync_playwright() exits, it closes the event loop, and trying to restart it in subsequent iterations will cause the error.
from src.services.LoggingService import LoggingService
from playwright.sync_api import sync_playwright, TimeoutError
from decouple import config
class UserService:
@staticmethod
def update(username, password, playerName, playerPassword):
max_attempts = 8
attempt = 0
with sync_playwright() as p: # Initialize Playwright ONCE
while attempt < max_attempts:
browser = None
try:
# Renew Proxy on each attempt
if config("PROXY_ENABLED") == "true":
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
proxy={"server": config("PROXY_URL")}
)
LoggingService.info(f"Proxy enabled: {config('PROXY_URL')}", "UserService")
else:
LoggingService.info("Proxy disabled", "UserService")
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(config("GA_LOGIN_URL"), timeout=60000)
page.fill('input.input_sizable_low.input_type_text.input_validState_1', username)
page.fill('input.input_sizable_low.input_type_password.input_with-eye.input_validState_1', password)
page.click('button.button_sizable_default.button_colors_default')
page.wait_for_selector('input.input_sizable_default.input_type_text.input_validState_1', timeout=60000)
page.fill('input.input_sizable_default.input_type_text.input_validState_1', playerName)
page.locator(f"text={playerName}").click(timeout=30000)
page.locator(".dropdown__action").first.click(timeout=30000)
page.get_by_role("link", name="Cambiar contraseña").click(timeout=30000)
page.get_by_role("textbox", name="Nueva contraseña", exact=True).fill(playerPassword)
page.get_by_role("textbox", name="Confirmar nueva contraseña").fill(playerPassword)
page.get_by_role("button", name="Cambiar").click(timeout=30000)
page.wait_for_timeout(1000)
browser.close() # Properly close the browser
LoggingService.info(f"Usuario {playerName} actualizado con éxito luego de {attempt + 1} intentos", "UserService")
return True
except Exception as e:
LoggingService.error(f"Error en intento {attempt + 1}: {str(e)}", "UserService")
attempt += 1
if browser:
try:
browser.close() # Close the browser before retrying
except:
pass # Ignore errors if the browser is already closed
if attempt == max_attempts:
LoggingService.error(f"No se pudo realizar el ingreso para {playerName} tras {max_attempts} intentos", "UserService")
return False
return False
Upvotes: 0
ggorlen
Reputation: 57195
The problem is not obviously reproducible from the code here. Please show how you're calling these static class methods, and remove any code that isn't relevant to the reproduction (probably one method is sufficient, and the method can simply navigate to an example site).
There is no problem with running with in a loop or not:
from playwright.sync_api import sync_playwright # 1.48.0
for _ in range(3):
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto("https://www.example.com")
print(page.title())
with sync_playwright() as p:
for _ in range(3):
browser = p.chromium.launch()
page = browser.new_page()
page.goto("https://www.example.com")
print(page.title())
This logs Example Domain 6 times and exits, as expected. Adding this into a class method doesn't change this:
class Foo:
@staticmethod
def bar():
for _ in range(3):
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto("https://www.example.com")
print(page.title())
with sync_playwright() as p:
for _ in range(3):
browser = p.chromium.launch()
page = browser.new_page()
page.goto("https://www.example.com")
print(page.title())
Foo().bar()
See this GH issue for a minimal reproduction of the titular error.
Upvotes: 0
Related Questions
- How can I install packages using pip according to the requirements.txt file from a local directory?
- How can I capitalize the first letter of each word in a string?
- How can I flush the output of the print function?
- How to access the index value in a 'for' loop?
- Obtain href links from first table in headless browser page (playwright._impl._errors.Error: Event loop is closed! Is Playwright already stopped?)
- How can I import a module dynamically given the full path?
- How can the Euclidean distance be calculated with NumPy?
- How can I get the source code of a Python function?
- How can I use Python to get the system hostname?