mirror of
https://github.com/FlareSolverr/FlareSolverr.git
synced 2025-06-08 12:35:30 +00:00
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Tuple
|
|
from uuid import uuid1
|
|
|
|
from selenium.webdriver.chrome.webdriver import WebDriver
|
|
|
|
import utils
|
|
|
|
|
|
@dataclass
|
|
class Session:
|
|
session_id: str
|
|
driver: WebDriver
|
|
created_at: datetime
|
|
|
|
def lifetime(self) -> timedelta:
|
|
return datetime.now() - self.created_at
|
|
|
|
|
|
class SessionsStorage:
|
|
"""SessionsStorage creates, stores and process all the sessions"""
|
|
|
|
def __init__(self):
|
|
self.sessions = {}
|
|
|
|
def create(self, session_id: Optional[str] = None, force_new: Optional[bool] = False) -> Tuple[Session, bool]:
|
|
"""create creates new instance of WebDriver if necessary,
|
|
assign defined (or newly generated) session_id to the instance
|
|
and returns the session object. If a new session has been created
|
|
second argument is set to True.
|
|
|
|
Note: The function is idempotent, so in case if session_id
|
|
already exists in the storage a new instance of WebDriver won't be created
|
|
and existing session will be returned. Second argument defines if
|
|
new session has been created (True) or an existing one was used (False).
|
|
"""
|
|
session_id = session_id or str(uuid1())
|
|
|
|
if force_new:
|
|
self.destroy(session_id)
|
|
|
|
if self.exists(session_id):
|
|
return self.sessions[session_id], False
|
|
|
|
driver = utils.get_webdriver()
|
|
created_at = datetime.now()
|
|
session = Session(session_id, driver, created_at)
|
|
|
|
self.sessions[session_id] = session
|
|
|
|
return session, True
|
|
|
|
def exists(self, session_id: str) -> bool:
|
|
return session_id in self.sessions
|
|
|
|
def destroy(self, session_id: str) -> bool:
|
|
"""destroy closes the driver instance and removes session from the storage.
|
|
The function is noop if session_id doesn't exist.
|
|
The function returns True if session was found and destroyed,
|
|
and False if session_id wasn't found.
|
|
"""
|
|
if not self.exists(session_id):
|
|
return False
|
|
|
|
session = self.sessions.pop(session_id)
|
|
session.driver.quit()
|
|
return True
|
|
|
|
def get(self, session_id: str, ttl: Optional[timedelta] = None) -> Tuple[Session, bool]:
|
|
session, fresh = self.create(session_id)
|
|
|
|
if ttl is not None and not fresh and session.lifetime() > ttl:
|
|
logging.debug(f'session\'s lifetime has expired, so the session is recreated (session_id={session_id})')
|
|
session, fresh = self.create(session_id, force_new=True)
|
|
|
|
return session, fresh
|
|
|
|
def session_ids(self) -> list[str]:
|
|
return list(self.sessions.keys())
|