跳转至

API reference

Generated from the apogee-ai-browser source with mkdocstrings. Every symbol below is exported from apogee_ai_browser, so it is part of the supported public surface.

Application · DTOs

BenchDTO dataclass

Python
BenchDTO(pages: int = 50, fixtures: dict[str, str] = dict())

pages class-attribute instance-attribute

Python
pages: int = 50

fixtures class-attribute instance-attribute

Python
fixtures: dict[str, str] = field(default_factory=dict)

CrawlDTO dataclass

Python
CrawlDTO(seed_url: str, max_depth: int = 2, max_pages: int = 10, allowed_domains: tuple[str, ...] = (), browser: str = 'stub')

seed_url instance-attribute

Python
seed_url: str

max_depth class-attribute instance-attribute

Python
max_depth: int = 2

max_pages class-attribute instance-attribute

Python
max_pages: int = 10

allowed_domains class-attribute instance-attribute

Python
allowed_domains: tuple[str, ...] = ()

browser class-attribute instance-attribute

Python
browser: str = 'stub'

ExtractDTO dataclass

Python
ExtractDTO(url: str, selector: str | None = None, browser: str = 'stub')

url instance-attribute

Python
url: str

selector class-attribute instance-attribute

Python
selector: str | None = None

browser class-attribute instance-attribute

Python
browser: str = 'stub'

NavigateDTO dataclass

Python
NavigateDTO(url: str, browser: str = 'stub')

url instance-attribute

Python
url: str

browser class-attribute instance-attribute

Python
browser: str = 'stub'

ScreenshotDTO dataclass

Python
ScreenshotDTO(url: str, output_path: str, browser: str = 'stub')

url instance-attribute

Python
url: str

output_path instance-attribute

Python
output_path: str

browser class-attribute instance-attribute

Python
browser: str = 'stub'

Application · Use cases

BenchNavigationUseCase

Synthetic crawl bench against StubBrowser fixtures.

execute async

Python
execute(pages: int) -> dict[str, float]
Source code in apogee_ai_browser/application/use_cases/bench_navigation_use_case.py
Python
async def execute(self, pages: int) -> dict[str, float]:
    if pages <= 0:
        raise ValueError("pages must be positive")
    fixtures = _synthetic_fixtures(pages)
    browser = StubBrowser(fixtures=fixtures)
    policy = NavigationPolicy(
        allowed_domains=("example.test",),
        max_depth=pages + 1,
        max_pages=pages,
    )
    crawl = CrawlUseCase(browser, policy)
    start = time.perf_counter()
    crawled = await crawl.execute("https://example.test/page-0000")
    elapsed = (time.perf_counter() - start) * 1000.0
    return {
        "pages_target": float(pages),
        "pages_crawled": float(len(crawled)),
        "elapsed_ms": elapsed,
        "pages_per_second": (len(crawled) / elapsed * 1000.0) if elapsed > 0 else 0.0,
    }

CrawlUseCase

Python
CrawlUseCase(browser, policy: NavigationPolicy, rate_limiter=None, robots_checker=None)

Bounded BFS crawl. Respects allowlist + max_depth + max_pages.

rate_limiter and robots_checker are optional. When supplied, every URL is checked before fetching.

Source code in apogee_ai_browser/application/use_cases/crawl_use_case.py
Python
def __init__(
    self,
    browser,
    policy: NavigationPolicy,
    rate_limiter=None,
    robots_checker=None,
) -> None:
    self._browser = browser
    self._policy = policy
    self._rate_limiter = rate_limiter
    self._robots = robots_checker

execute async

Python
execute(seed_url: str) -> list[Page]
Source code in apogee_ai_browser/application/use_cases/crawl_use_case.py
Python
async def execute(self, seed_url: str) -> list[Page]:
    if not self._policy.is_allowed(seed_url):
        raise PolicyViolationException(seed_url, "seed not in allowlist")

    await self._browser.open()
    visited: set[str] = set()
    queue: deque[tuple[str, int]] = deque([(seed_url, 0)])
    pages: list[Page] = []

    while queue and len(pages) < self._policy.max_pages:
        url, depth = queue.popleft()
        if url in visited or depth > self._policy.max_depth:
            continue
        visited.add(url)

        if not self._policy.is_allowed(url):
            continue
        if self._robots is not None and not await self._robots.can_fetch(
            url, self._policy.user_agent
        ):
            continue
        host = self._policy.host_of(url)
        if self._rate_limiter is not None and not await self._rate_limiter.acquire(host):
            raise RateLimitedException(host)

        try:
            page = await self._browser.navigate(url)
        except NavigationError:
            continue
        pages.append(page)

        if depth >= self._policy.max_depth:
            continue
        for href in _LINK_RE.findall(page.html or ""):
            target = urljoin(url, href)
            if target not in visited:
                queue.append((target, depth + 1))

    return pages

ExtractTextUseCase

Python
ExtractTextUseCase(browser)
Source code in apogee_ai_browser/application/use_cases/extract_text_use_case.py
Python
def __init__(self, browser) -> None:
    self._browser = browser

execute async

Python
execute(url: str, selector: ElementSelector | None = None) -> str
Source code in apogee_ai_browser/application/use_cases/extract_text_use_case.py
Python
async def execute(self, url: str, selector: ElementSelector | None = None) -> str:
    await self._browser.open()
    page = await self._browser.navigate(url)
    if selector is None:
        return page.text
    element = await self._browser.find(selector)
    return element.text

NavigateUseCase

Python
NavigateUseCase(browser, policy=None)
Source code in apogee_ai_browser/application/use_cases/navigate_use_case.py
Python
def __init__(self, browser, policy=None) -> None:
    self._browser = browser
    self._policy = policy

execute async

Python
execute(url: str) -> Page
Source code in apogee_ai_browser/application/use_cases/navigate_use_case.py
Python
async def execute(self, url: str) -> Page:
    if self._policy is not None and not await self._policy.is_allowed(url):
        from ...domain.exceptions.browser_exceptions import (
            PolicyViolationException,
        )

        raise PolicyViolationException(url, "domain not allowed")
    await self._browser.open()
    return await self._browser.navigate(url)

ScreenshotUseCase

Python
ScreenshotUseCase(browser)
Source code in apogee_ai_browser/application/use_cases/screenshot_use_case.py
Python
def __init__(self, browser) -> None:
    self._browser = browser

execute async

Python
execute(url: str, output_path: str) -> str
Source code in apogee_ai_browser/application/use_cases/screenshot_use_case.py
Python
async def execute(self, url: str, output_path: str) -> str:
    await self._browser.open()
    await self._browser.navigate(url)
    return await self._browser.screenshot(output_path)

Domain

BrowserAction dataclass

Python
BrowserAction(kind: ActionKind, target: str | None = None, value: str | None = None)

kind instance-attribute

Python
kind: ActionKind

target class-attribute instance-attribute

Python
target: str | None = None

value class-attribute instance-attribute

Python
value: str | None = None

BrowserSession dataclass

Python
BrowserSession(id: str, user_agent: str = 'Apogee-AI-Browser/0.1', visited: set[str] = set(), cookies: dict[str, str] = dict(), current_url: str | None = None)

Mutable session state — visited urls, cookies, current page url.

id instance-attribute

Python
id: str

user_agent class-attribute instance-attribute

Python
user_agent: str = 'Apogee-AI-Browser/0.1'

visited class-attribute instance-attribute

Python
visited: set[str] = field(default_factory=set)

cookies class-attribute instance-attribute

Python
cookies: dict[str, str] = field(default_factory=dict)

current_url class-attribute instance-attribute

Python
current_url: str | None = None

mark_visited

Python
mark_visited(url: str) -> None
Source code in apogee_ai_browser/domain/entities/browser_session.py
Python
def mark_visited(self, url: str) -> None:
    self.visited.add(url)
    self.current_url = url

Element dataclass

Python
Element(selector: ElementSelector, text: str = '', attributes: dict[str, str] = dict(), tag: str = '')

selector instance-attribute

Python
selector: ElementSelector

text class-attribute instance-attribute

Python
text: str = ''

attributes class-attribute instance-attribute

Python
attributes: dict[str, str] = field(default_factory=dict)

tag class-attribute instance-attribute

Python
tag: str = ''

ElementSelector dataclass

Python
ElementSelector(kind: SelectorKind, value: str)

kind instance-attribute

Python
kind: SelectorKind

value instance-attribute

Python
value: str

css classmethod

Python
css(value: str) -> 'ElementSelector'
Source code in apogee_ai_browser/domain/value_objects/element_selector.py
Python
@classmethod
def css(cls, value: str) -> "ElementSelector":
    return cls(kind=SelectorKind.CSS, value=value)

xpath classmethod

Python
xpath(value: str) -> 'ElementSelector'
Source code in apogee_ai_browser/domain/value_objects/element_selector.py
Python
@classmethod
def xpath(cls, value: str) -> "ElementSelector":
    return cls(kind=SelectorKind.XPATH, value=value)

text classmethod

Python
text(value: str) -> 'ElementSelector'
Source code in apogee_ai_browser/domain/value_objects/element_selector.py
Python
@classmethod
def text(cls, value: str) -> "ElementSelector":
    return cls(kind=SelectorKind.TEXT, value=value)

NavigationPolicy dataclass

Python
NavigationPolicy(allowed_domains: tuple[str, ...] = (), blocked_domains: tuple[str, ...] = (), max_depth: int = 3, max_pages: int = 50, respect_robots: bool = True, user_agent: str = 'Apogee-AI-Browser/0.1', request_timeout_s: float = 10.0, requests_per_minute_per_host: int = 30, metadata: dict[str, str] = dict())

allowed_domains class-attribute instance-attribute

Python
allowed_domains: tuple[str, ...] = ()

blocked_domains class-attribute instance-attribute

Python
blocked_domains: tuple[str, ...] = ()

max_depth class-attribute instance-attribute

Python
max_depth: int = 3

max_pages class-attribute instance-attribute

Python
max_pages: int = 50

respect_robots class-attribute instance-attribute

Python
respect_robots: bool = True

user_agent class-attribute instance-attribute

Python
user_agent: str = 'Apogee-AI-Browser/0.1'

request_timeout_s class-attribute instance-attribute

Python
request_timeout_s: float = 10.0

requests_per_minute_per_host class-attribute instance-attribute

Python
requests_per_minute_per_host: int = 30

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

host_of

Python
host_of(url: str) -> str
Source code in apogee_ai_browser/domain/value_objects/navigation_policy.py
Python
def host_of(self, url: str) -> str:
    return (urlparse(url).hostname or "").lower()

is_allowed

Python
is_allowed(url: str) -> bool
Source code in apogee_ai_browser/domain/value_objects/navigation_policy.py
Python
def is_allowed(self, url: str) -> bool:
    host = self.host_of(url)
    if not host:
        return False
    if any(host == bad or host.endswith("." + bad) for bad in self.blocked_domains):
        return False
    if not self.allowed_domains:
        return True
    return any(
        host == allow or host.endswith("." + allow)
        for allow in self.allowed_domains
    )

NavigationStep dataclass

Python
NavigationStep(kind: ActionKind, url: str, depth: int = 0, parent_url: str | None = None, timestamp_s: float = 0.0)

kind instance-attribute

Python
kind: ActionKind

url instance-attribute

Python
url: str

depth class-attribute instance-attribute

Python
depth: int = 0

parent_url class-attribute instance-attribute

Python
parent_url: str | None = None

timestamp_s class-attribute instance-attribute

Python
timestamp_s: float = 0.0

Page dataclass

Python
Page(url: str, title: str = '', text: str = '', html: str = '', status: int = 200, screenshot_path: str | None = None, headers: dict[str, str] = dict(), metadata: dict[str, str] = dict())

url instance-attribute

Python
url: str

title class-attribute instance-attribute

Python
title: str = ''

text class-attribute instance-attribute

Python
text: str = ''

html class-attribute instance-attribute

Python
html: str = ''

status class-attribute instance-attribute

Python
status: int = 200

screenshot_path class-attribute instance-attribute

Python
screenshot_path: str | None = None

headers class-attribute instance-attribute

Python
headers: dict[str, str] = field(default_factory=dict)

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

Domain · Enums

ActionKind

Bases: str, Enum

NAVIGATE class-attribute instance-attribute

Python
NAVIGATE = 'navigate'

CLICK class-attribute instance-attribute

Python
CLICK = 'click'

TYPE class-attribute instance-attribute

Python
TYPE = 'type'

SCREENSHOT class-attribute instance-attribute

Python
SCREENSHOT = 'screenshot'

EXTRACT_TEXT class-attribute instance-attribute

Python
EXTRACT_TEXT = 'extract_text'

WAIT class-attribute instance-attribute

Python
WAIT = 'wait'

BrowserKind

Bases: str, Enum

STUB class-attribute instance-attribute

Python
STUB = 'stub'

HTTP_FETCH class-attribute instance-attribute

Python
HTTP_FETCH = 'http_fetch'

PLAYWRIGHT class-attribute instance-attribute

Python
PLAYWRIGHT = 'playwright'

SelectorKind

Bases: str, Enum

CSS class-attribute instance-attribute

Python
CSS = 'css'

XPATH class-attribute instance-attribute

Python
XPATH = 'xpath'

TEXT class-attribute instance-attribute

Python
TEXT = 'text'

ROLE class-attribute instance-attribute

Python
ROLE = 'role'

Domain · Exceptions

BrowserError

Bases: Exception

Base for apogee-ai-browser errors.

ElementNotFoundException

Python
ElementNotFoundException(selector: str)

Bases: BrowserError

Source code in apogee_ai_browser/domain/exceptions/browser_exceptions.py
Python
def __init__(self, selector: str) -> None:
    super().__init__(f"Element not found: {selector}")
    self.selector = selector

selector instance-attribute

Python
selector = selector

NavigationError

Python
NavigationError(url: str, message: str)

Bases: BrowserError

Source code in apogee_ai_browser/domain/exceptions/browser_exceptions.py
Python
def __init__(self, url: str, message: str) -> None:
    super().__init__(f"Navigation to {url!r} failed: {message}")
    self.url = url

url instance-attribute

Python
url = url

PolicyViolationException

Python
PolicyViolationException(url: str, reason: str)

Bases: BrowserError

Source code in apogee_ai_browser/domain/exceptions/browser_exceptions.py
Python
def __init__(self, url: str, reason: str) -> None:
    super().__init__(f"Policy blocked {url}: {reason}")
    self.url = url
    self.reason = reason

url instance-attribute

Python
url = url

reason instance-attribute

Python
reason = reason

RateLimitedException

Python
RateLimitedException(host: str)

Bases: BrowserError

Source code in apogee_ai_browser/domain/exceptions/browser_exceptions.py
Python
def __init__(self, host: str) -> None:
    super().__init__(f"Rate-limited for host {host!r}")
    self.host = host

host instance-attribute

Python
host = host

Domain · Protocols (ports)

IBrowser

Bases: Protocol

name instance-attribute

Python
name: str

open async

Python
open() -> None
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def open(self) -> None: ...

navigate async

Python
navigate(url: str) -> Page
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def navigate(self, url: str) -> Page: ...

find async

Python
find(selector: ElementSelector) -> Element
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def find(self, selector: ElementSelector) -> Element: ...

click async

Python
click(selector: ElementSelector) -> None
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def click(self, selector: ElementSelector) -> None: ...

type_text async

Python
type_text(selector: ElementSelector, value: str) -> None
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def type_text(self, selector: ElementSelector, value: str) -> None: ...

screenshot async

Python
screenshot(path: str) -> str
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def screenshot(self, path: str) -> str: ...

close async

Python
close() -> None
Source code in apogee_ai_browser/domain/services/i_browser.py
Python
async def close(self) -> None: ...

INavigationPolicyCheck

Bases: Protocol

is_allowed async

Python
is_allowed(url: str) -> bool
Source code in apogee_ai_browser/domain/services/i_navigation_policy_check.py
Python
async def is_allowed(self, url: str) -> bool: ...

IRateLimiter

Bases: Protocol

acquire async

Python
acquire(host: str) -> bool
Source code in apogee_ai_browser/domain/services/i_rate_limiter.py
Python
async def acquire(self, host: str) -> bool: ...

IRobotsChecker

Bases: Protocol

can_fetch async

Python
can_fetch(url: str, user_agent: str) -> bool
Source code in apogee_ai_browser/domain/services/i_robots_checker.py
Python
async def can_fetch(self, url: str, user_agent: str) -> bool: ...

Infrastructure

AllowAllRobotsChecker

Default checker — bypasses robots.txt entirely (CI / tests).

can_fetch async

Python
can_fetch(url: str, user_agent: str) -> bool
Source code in apogee_ai_browser/infrastructure/policies/robots_checker.py
Python
async def can_fetch(self, url: str, user_agent: str) -> bool:
    return True

DomainAllowlistPolicy

Python
DomainAllowlistPolicy(policy: NavigationPolicy)

Wraps NavigationPolicy.is_allowed in an async-friendly check.

Source code in apogee_ai_browser/infrastructure/policies/domain_allowlist.py
Python
def __init__(self, policy: NavigationPolicy) -> None:
    self._policy = policy

is_allowed async

Python
is_allowed(url: str) -> bool
Source code in apogee_ai_browser/infrastructure/policies/domain_allowlist.py
Python
async def is_allowed(self, url: str) -> bool:
    return self._policy.is_allowed(url)

HttpFetchBrowser

Python
HttpFetchBrowser(timeout_s: float = 10.0, user_agent: str = 'Apogee-AI-Browser/0.1')

Static HTTP fetch via urllib. No JS, no cookies persistence.

Always available — no extra dependency required.

Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
def __init__(
    self,
    timeout_s: float = 10.0,
    user_agent: str = "Apogee-AI-Browser/0.1",
) -> None:
    self._timeout = timeout_s
    self._user_agent = user_agent
    self._current: Page | None = None

name class-attribute instance-attribute

Python
name = 'http_fetch'

open async

Python
open() -> None
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def open(self) -> None:  # nothing to do for stateless fetch
    return None

navigate async

Python
navigate(url: str) -> Page
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def navigate(self, url: str) -> Page:
    status, headers, html = await asyncio.to_thread(self._fetch_sync, url)
    page = Page(
        url=url,
        title=_extract_title(html),
        text=_extract_text(html),
        html=html,
        status=status,
        headers={k.lower(): v for k, v in headers.items()},
        metadata={"links": ",".join(_LINK_RE.findall(html))},
    )
    self._current = page
    return page

find async

Python
find(selector: ElementSelector) -> Element
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def find(self, selector: ElementSelector) -> Element:
    if self._current is None or selector.value not in self._current.text:
        raise ElementNotFoundException(str(selector))
    return Element(selector=selector, text=selector.value)

click async

Python
click(selector: ElementSelector) -> None
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def click(self, selector: ElementSelector) -> None:
    raise NotImplementedError("HttpFetchBrowser is read-only; use Playwright")

type_text async

Python
type_text(selector: ElementSelector, value: str) -> None
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def type_text(self, selector: ElementSelector, value: str) -> None:
    raise NotImplementedError("HttpFetchBrowser is read-only; use Playwright")

screenshot async

Python
screenshot(path: str) -> str
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def screenshot(self, path: str) -> str:
    raise NotImplementedError("HttpFetchBrowser cannot render — use Playwright")

close async

Python
close() -> None
Source code in apogee_ai_browser/infrastructure/browsers/http_fetch_browser.py
Python
async def close(self) -> None:
    self._current = None

InMemorySessionStore

Python
InMemorySessionStore()

Tracks browser sessions by id for the lifetime of a process.

Source code in apogee_ai_browser/infrastructure/sessions/in_memory_session_store.py
Python
def __init__(self) -> None:
    self._sessions: dict[str, BrowserSession] = {}

create

Python
create(session_id: str, user_agent: str = 'Apogee-AI-Browser/0.1') -> BrowserSession
Source code in apogee_ai_browser/infrastructure/sessions/in_memory_session_store.py
Python
def create(self, session_id: str, user_agent: str = "Apogee-AI-Browser/0.1") -> BrowserSession:
    if session_id in self._sessions:
        return self._sessions[session_id]
    session = BrowserSession(id=session_id, user_agent=user_agent)
    self._sessions[session_id] = session
    return session

get

Python
get(session_id: str) -> BrowserSession | None
Source code in apogee_ai_browser/infrastructure/sessions/in_memory_session_store.py
Python
def get(self, session_id: str) -> BrowserSession | None:
    return self._sessions.get(session_id)

drop

Python
drop(session_id: str) -> None
Source code in apogee_ai_browser/infrastructure/sessions/in_memory_session_store.py
Python
def drop(self, session_id: str) -> None:
    self._sessions.pop(session_id, None)

count

Python
count() -> int
Source code in apogee_ai_browser/infrastructure/sessions/in_memory_session_store.py
Python
def count(self) -> int:
    return len(self._sessions)

PerHostRateLimiter

Python
PerHostRateLimiter(requests_per_minute: int = 30, burst: int = 1)

Token bucket per host. Refills linearly at requests_per_minute / 60.

Source code in apogee_ai_browser/infrastructure/policies/per_host_rate_limiter.py
Python
def __init__(self, requests_per_minute: int = 30, burst: int = 1) -> None:
    if requests_per_minute <= 0:
        raise ValueError("requests_per_minute must be positive")
    if burst <= 0:
        raise ValueError("burst must be positive")
    self._rpm = requests_per_minute
    self._capacity = float(requests_per_minute * burst)
    self._buckets: dict[str, _Bucket] = {}

acquire async

Python
acquire(host: str) -> bool
Source code in apogee_ai_browser/infrastructure/policies/per_host_rate_limiter.py
Python
async def acquire(self, host: str) -> bool:
    b = self._bucket(host)
    if b.tokens >= 1.0:
        b.tokens -= 1.0
        return True
    return False

PlaywrightBrowser

Python
PlaywrightBrowser(headless: bool = True, engine: str = 'chromium', user_agent: str = 'Apogee-AI-Browser/0.1')

Lazy Playwright adapter — install via extras=playwright.

Supports full JS rendering, click, type, screenshot.

Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
def __init__(
    self,
    headless: bool = True,
    engine: str = "chromium",
    user_agent: str = "Apogee-AI-Browser/0.1",
) -> None:
    self._headless = headless
    self._engine = engine
    self._user_agent = user_agent
    self._playwright = None
    self._browser = None
    self._context = None
    self._page = None

name class-attribute instance-attribute

Python
name = 'playwright'

open async

Python
open() -> None
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def open(self) -> None:
    if self._page is not None:
        return
    try:
        from playwright.async_api import async_playwright  # type: ignore
    except ImportError as exc:  # pragma: no cover
        raise BrowserError(
            "install with `pip install apogee-ai-browser[playwright]` "
            "and run `playwright install chromium`"
        ) from exc
    self._playwright = await async_playwright().start()
    engine = getattr(self._playwright, self._engine)
    self._browser = await engine.launch(headless=self._headless)
    self._context = await self._browser.new_context(user_agent=self._user_agent)
    self._page = await self._context.new_page()

navigate async

Python
navigate(url: str) -> Page
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def navigate(self, url: str) -> Page:
    await self.open()
    try:
        response = await self._page.goto(url)  # type: ignore[union-attr]
    except Exception as exc:  # pragma: no cover - depends on Playwright
        raise NavigationError(url, str(exc)) from exc
    title = await self._page.title()  # type: ignore[union-attr]
    text = await self._page.evaluate("() => document.body.innerText || ''")  # type: ignore[union-attr]
    html = await self._page.content()  # type: ignore[union-attr]
    return Page(
        url=url,
        title=title or "",
        text=text or "",
        html=html or "",
        status=response.status if response else 0,
    )

find async

Python
find(selector: ElementSelector) -> Element
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def find(self, selector: ElementSelector) -> Element:
    locator = self._to_locator(selector)
    if await locator.count() == 0:
        raise ElementNotFoundException(str(selector))
    text = await locator.first.text_content()
    return Element(selector=selector, text=text or "")

click async

Python
click(selector: ElementSelector) -> None
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def click(self, selector: ElementSelector) -> None:
    await self._to_locator(selector).first.click()

type_text async

Python
type_text(selector: ElementSelector, value: str) -> None
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def type_text(self, selector: ElementSelector, value: str) -> None:
    await self._to_locator(selector).first.fill(value)

screenshot async

Python
screenshot(path: str) -> str
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def screenshot(self, path: str) -> str:
    from pathlib import Path

    Path(path).parent.mkdir(parents=True, exist_ok=True)
    await self._page.screenshot(path=path)  # type: ignore[union-attr]
    return path

close async

Python
close() -> None
Source code in apogee_ai_browser/infrastructure/browsers/playwright_browser.py
Python
async def close(self) -> None:
    if self._context is not None:
        await self._context.close()
        self._context = None
    if self._browser is not None:
        await self._browser.close()
        self._browser = None
    if self._playwright is not None:
        await self._playwright.stop()
        self._playwright = None
    self._page = None

RobotsTxtChecker

Python
RobotsTxtChecker(timeout_s: float = 5.0)

Caches robots.txt per host. Defaults to allow on fetch failure.

Source code in apogee_ai_browser/infrastructure/policies/robots_checker.py
Python
def __init__(self, timeout_s: float = 5.0) -> None:
    self._timeout = timeout_s
    self._cache: dict[str, RobotFileParser] = {}

can_fetch async

Python
can_fetch(url: str, user_agent: str) -> bool
Source code in apogee_ai_browser/infrastructure/policies/robots_checker.py
Python
async def can_fetch(self, url: str, user_agent: str) -> bool:
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https") or not parsed.hostname:
        return False
    host_key = f"{parsed.scheme}://{parsed.hostname}"
    if host_key not in self._cache:
        self._cache[host_key] = await asyncio.to_thread(
            self._fetch_sync, f"{host_key}/robots.txt"
        )
    return self._cache[host_key].can_fetch(user_agent, url)

StubBrowser

Python
StubBrowser(fixtures: dict[str, str] | None = None)

Deterministic in-memory browser for tests / CI. No network.

Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
def __init__(self, fixtures: dict[str, str] | None = None) -> None:
    self._fixtures = dict(fixtures or {})
    self._open = False
    self._current: Page | None = None

name class-attribute instance-attribute

Python
name = 'stub'

add_fixture

Python
add_fixture(url: str, html: str) -> None
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
def add_fixture(self, url: str, html: str) -> None:
    self._fixtures[url] = html

open async

Python
open() -> None
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def open(self) -> None:
    self._open = True

navigate async

Python
navigate(url: str) -> Page
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def navigate(self, url: str) -> Page:
    if not self._open:
        await self.open()
    if url not in self._fixtures:
        raise NavigationError(url, "no fixture registered")
    html = self._fixtures[url]
    page = Page(
        url=url,
        title=_extract_title(html),
        text=_extract_text(html),
        html=html,
        status=200,
        headers={"content-type": "text/html"},
        metadata={"links": ",".join(_LINK_RE.findall(html))},
    )
    self._current = page
    return page

find async

Python
find(selector: ElementSelector) -> Element
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def find(self, selector: ElementSelector) -> Element:
    if self._current is None:
        raise ElementNotFoundException(str(selector))
    if selector.value in self._current.text:
        return Element(selector=selector, text=selector.value)
    raise ElementNotFoundException(str(selector))

click async

Python
click(selector: ElementSelector) -> None
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def click(self, selector: ElementSelector) -> None:
    await self.find(selector)

type_text async

Python
type_text(selector: ElementSelector, value: str) -> None
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def type_text(self, selector: ElementSelector, value: str) -> None:
    await self.find(selector)

screenshot async

Python
screenshot(path: str) -> str
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def screenshot(self, path: str) -> str:
    from pathlib import Path

    out = Path(path)
    out.parent.mkdir(parents=True, exist_ok=True)
    url = self._current.url if self._current else "about:blank"
    out.write_text(f"# stub screenshot of {url}\n", encoding="utf-8")
    return str(out)

close async

Python
close() -> None
Source code in apogee_ai_browser/infrastructure/browsers/stub_browser.py
Python
async def close(self) -> None:
    self._open = False
    self._current = None