need to test singleton

This commit is contained in:
2022-07-25 14:21:12 +02:00
parent 3160fefdab
commit f482164c16
4 changed files with 48 additions and 17 deletions
+1 -6
View File
@@ -17,6 +17,7 @@ from src.pojo.ReserveResultPojo import ReserveResultPojo
from src.pojo.ResultEnum import ResultEnum
from src.proxy.proxy_type import ProxyType
from src import definitions, params
from src.workers.TlsPlaywright import TlsPlaywright
SORRY_SENTENCE_FR = "nous sommes sincèrement désolés de n'avoir pu vous satisfaire cette fois-ci"
SORRY_SENTENCE_EN = "we are extremely sorry that we were not able to fulfill"
@@ -31,12 +32,6 @@ mailer = Mailer()
oracle_log_sender = LogSender()
class TlsPlaywright(threading.local):
def __init__(self) -> None:
self.playwright = sync_playwright().start()
print("创建浏览器实例,线程: ", threading.current_thread().name)
class ResultChecker:
tls = TlsPlaywright()
+40
View File
@@ -0,0 +1,40 @@
import threading
from playwright.sync_api import sync_playwright
class OnlyOne:
class __OnlyOne:
def __init__(self, arg):
self.val = arg
def __str__(self):
return repr(self) + self.val
instance = None
def __init__(self, arg):
if not OnlyOne.instance:
OnlyOne.instance = OnlyOne.__OnlyOne(arg)
else:
OnlyOne.instance.val = arg
def __getattr__(self, name):
return getattr(self.instance, name)
class TlsPlaywright():
# private nested inner class
class __TlsPlaywrightSingleton(threading.local):
def __init__(self) -> None:
self.playwright = sync_playwright().start()
print("创建浏览器实例,线程: ", threading.current_thread().name)
playwright = None
def __init__(self):
if not TlsPlaywright.playwright:
TlsPlaywright.playwright = TlsPlaywright.__TlsPlaywrightSingleton().playwright
def __getattr__(self, name):
return getattr(self.playwright, name)
+2 -6
View File
@@ -17,6 +17,7 @@ from src.pojo.ReserveResultPojo import ReserveResultPojo, PublishType
from src.pojo.contact_pojo import ContactPojo
from src.proxy.proxy_type import ProxyType
from src.workers.SolveCaptch import SolveCaptcha
from src.workers.TlsPlaywright import TlsPlaywright
RDV_URL = "https://rendezvousparis.hermes.com/client/register"
@@ -46,13 +47,8 @@ def get_random_wait_time() -> float:
return wait_time
class Tls(threading.local):
def __init__(self) -> None:
self.playwright = sync_playwright().start()
class CommandorPage:
tls = Tls()
tls = TlsPlaywright()
def __init__(self, contact: ContactPojo, store_type=0, proxy_type=ProxyType.BRIGHT_DATA,
mode: ModeEnum = ModeEnum.MANUAL, headless=False):