Advanced Topics¶

from xeepy.core.browser import BrowserManager

class BrowserManager:
    """Manages browser lifecycle and page pool."""

    async def start(self) -> None:
        """Launch browser with stealth configuration."""

    async def new_page(self) -> Page:
        """Create new page with rate limiting."""

    async def close(self) -> None:
        """Clean shutdown with session save."""

from xeepy.core.rate_limiter import RateLimiter

class RateLimiter:
    """Adaptive rate limiter with backoff."""

    def __init__(
        self,
        requests_per_minute: int = 20,
        burst_limit: int = 5,
        backoff_factor: float = 2.0
    ): ...

    async def wait(self) -> None:
        """Wait for rate limit clearance."""

    def record_response(self, status: int) -> None:
        """Adjust limits based on response."""

from xeepy.core.events import EventEmitter

class Xeepy(EventEmitter):
    """Emits events during operations."""

    # Events:
    # - "scrape:start", "scrape:complete", "scrape:error"
    # - "action:start", "action:complete", "action:error"
    # - "auth:login", "auth:logout", "auth:expired"
    # - "rate_limit:warning", "rate_limit:hit"

# Subscribe to events
x.on("scrape:complete", lambda data: print(f"Scraped {len(data)} items"))
x.on("rate_limit:warning", lambda: print("Slowing down..."))

from xeepy.scrapers.base import BaseScraper

class MyScraper(BaseScraper):
    """Custom scraper implementation."""

    async def scrape(self, target: str, **kwargs) -> ScrapeResult:
        # Your implementation
        pass

from xeepy.actions.base import BaseAction

class MyAction(BaseAction):
    """Custom action implementation."""

    async def execute(self, **kwargs) -> ActionResult:
        # Your implementation
        pass

from xeepy.notifications.base import BaseNotifier

class MyNotifier(BaseNotifier):
    """Custom notification channel."""

    async def send(self, message: str, **kwargs) -> bool:
        # Your implementation
        pass