Caching Http Requests

[vincentmuthomi]

Caching does not seem to work

My setup

• Chrome -> not chromium
• Persistent context
• A http proxy server
• No page.route or context.route. Saw that there was an issue with this in caching, but I do not implement it.

The Issue

When using an ordinary browser, I get 200 OK (from memory cache) and the x-cache: RefreshHit from cloudfront
On the patchright controlled environment, I get 200 OK but the x-cache header is non -existent in most staticly served files.

The result, the page takes forever to reload

Diagnosis

• Tried to run the proxy server across a normal browser without playwright and the result was as expected, caching was working fine hence ruling out the effects of the proxy server

• Tried to change a couple of default injected args when running chrome, nothing worked

Below is the setup code I used, also to add, I was testing the ignore-default-args one by one. The script below showcases all the arguments I tested.

 async with async_playwright() as p:        
                
                proxyConfig = self.proxyConfig
                host = proxyConfig.host if proxyConfig else "no proxy"
                port = proxyConfig.port if proxyConfig else "no proxy"
                context = await p.chromium.launch_persistent_context(
                    user_data_dir="chrome-data",
                headless=False,
                args=["--no-sandbox",
                    '--disable-blink-features=AutomationControlled',
                        "--disk-cache-size=104857600", 
                        "--memory-cache-size=52428800",  
                ],
                ignore_default_args=[
    "--disable-back-forward-cache",
    "--disable-field-trial-config",
    "--disable-background-networking",
    "--disable-features=AvoidUnnecessaryBeforeUnloadCheckSync,BoundaryEventDispatchTracksNodeRemoval,DestroyProfileOnBrowserClose,DialMediaRouteProvider,GlobalMediaControls,HttpsUpgrades,LensOverlay,MediaRouter,PaintHolding,ThirdPartyStoragePartitioning,Translate,AutoDeElevate,RenderDocument,OptimizationHints",
],
                channel="chrome",
                #executable_path=chrome_source(),
                no_viewport=True,
                accept_downloads=True,
                proxy={"server": f"http://{proxyConfig.host}:{proxyConfig.port}",
                       "username":proxyConfig.username,
                       "password":proxyConfig.password
                       } if proxyConfig else None
                )

My thought process

Something in either playwright or patchright may be causing the problem. Any assistance will be appreciated

[SakalOFF]

I just use

traffic_recorder.py

from future import annotations
import asyncio
import json
import re
import time
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional, Callable, Iterable, Union

from patchright.async_api import BrowserContext, Request as PwRequest, Response as PwResponse

_JSON = Union[dict, list, str, int, float, bool, None]

@DataClass
class RequestRecord:
ts: float
method: str
url: str
request_headers: Dict[str, str] = field(default_factory=dict)
request_body: Optional[str] = None
response_status: Optional[int] = None
response_headers: Dict[str, str] = field(default_factory=dict)
response_body: Optional[str] = None
response_json: Optional[_JSON] = None
page_url_at_time: Optional[str] = None # полезно при дебаге навигации

class RequestHistory:
def init(
self,
capture_bodies: bool = True,
max_body_bytes: int = 1_000_000,
decode: str = "utf-8",
):
self._records: List[RequestRecord] = []
self._by_req_id: Dict[str, RequestRecord] = {}
self._capture_bodies = capture_bodies
self._max_body_bytes = max_body_bytes
self._decode = decode
self._attached = False
self._lock = asyncio.Lock()

def clear(self):
    self._records.clear()
    self._by_req_id.clear()

def all(self) -> List[RequestRecord]:
    # возвращаем копию, чтобы не сломали из вне
    return list(self._records)

async def attach(self, context: BrowserContext):
    """
    Подписываемся на события контекста: request/response
    (ловит трафик со всех страниц).
    """
    if self._attached:
        return
    self._attached = True

    async def on_request(req: PwRequest):
        try:
            rid = req._impl_obj._channel._guid  # внутренний id (стабильно в пределах сессии)
        except Exception:
            rid = f"{id(req)}-{time.time()}"

        body_text: Optional[str] = None
        if self._capture_bodies:
            try:
                post_data = req.post_data
                if isinstance(post_data, (bytes, bytearray)):
                    body_text = post_data[: self._max_body_bytes].decode(self._decode, errors="replace")
                elif isinstance(post_data, str):
                    body_text = post_data[: self._max_body_bytes]
            except Exception:
                body_text = None

        page_url = None
        try:
            if req.frame and req.frame.page:
                page_url = req.frame.page.url
        except Exception:
            pass

        rec = RequestRecord(
            ts=time.time(),
            method=req.method,
            url=req.url,
            request_headers=await _lowered_headers(req.headers),
            request_body=body_text,
            page_url_at_time=page_url,
        )
        async with self._lock:
            self._records.append(rec)
            self._by_req_id[rid] = rec

    async def on_response(resp: PwResponse):
        try:
            req = resp.request
            try:
                rid = req._impl_obj._channel._guid
            except Exception:
                rid = None

            async with self._lock:
                rec = self._by_req_id.get(rid)
                # Если не поймали request (маловероятно), создадим базу
                if rec is None:
                    rec = RequestRecord(
                        ts=time.time(),
                        method=req.method,
                        url=req.url,
                        request_headers=await _lowered_headers(req.headers),
                    )
                    self._records.append(rec)

            rec.response_status = resp.status
            rec.response_headers = await _lowered_headers(resp.headers)

            if self._capture_bodies:
                try:
                    body_bytes = await resp.body()
                    if body_bytes:
                        body_bytes = body_bytes[: self._max_body_bytes]
                        rec.response_body = body_bytes.decode(self._decode, errors="replace")
                        # Попробуем JSON
                        try:
                            rec.response_json = json.loads(rec.response_body)
                        except Exception:
                            rec.response_json = None
                except Exception:
                    pass
        except Exception:
            # глотаем ошибки, чтобы не влияли на основной сценарий
            pass

    context.on("request", on_request)
    context.on("response", on_response)

# =============================
# Поисковые утилиты
# =============================

def search(
    self,
    url_contains: Optional[str] = None,
    url_regex: Optional[re.Pattern] = None,
    method: Optional[str] = None,
    request_headers_contains: Optional[Dict[str, str]] = None,
    response_headers_contains: Optional[Dict[str, str]] = None,
    response_status: Optional[int] = None,
    response_json_matches: Optional[Dict[str, Any]] = None,
    from_end: bool = False,
    limit: Optional[int] = None,
    custom_predicate: Optional[Callable[[RequestRecord], bool]] = None,
) -> List[RequestRecord]:
    """
    Ищет записи, удовлетворяющие ВСЕМ заданным критериям.
    - url_contains: подстрока (регистр нечувствителен).
    - url_regex: регулярка re.Pattern (если нужно сложнее).
    - method: GET/POST/...
    - request_headers_contains / response_headers_contains: подмножество (ключи сравниваются в lowercase).
    - response_status: например 200.
    - response_json_matches: { "a.b[0].c": "value", ... } — проверка полей JSON по дот-пути.
    - from_end: проходить историю с конца (последние совпадения первыми).
    - limit: ограничить количество результатов.
    - custom_predicate: своя функция (rec) -> bool.
    """
    seq: Iterable[RequestRecord] = reversed(self._records) if from_end else self._records

    out: List[RequestRecord] = []
    for rec in seq:
        if method and rec.method.upper() != method.upper():
            continue

        if url_contains and url_contains.lower() not in rec.url.lower():
            continue

        if url_regex and not url_regex.search(rec.url):
            continue

        if response_status is not None and rec.response_status != response_status:
            continue

        if request_headers_contains and not _headers_contains(rec.request_headers, request_headers_contains):
            continue

        if response_headers_contains and not _headers_contains(rec.response_headers, response_headers_contains):
            continue

        if response_json_matches:
            if rec.response_json is None:
                continue
            if not _json_matches(rec.response_json, response_json_matches):
                continue

        if custom_predicate and not custom_predicate(rec):
            continue

        out.append(rec)
        if limit is not None and len(out) >= limit:
            break

    return out

def search_one(
    self,
    **kwargs
) -> Optional[RequestRecord]:
    """
    Возвращает первую (или последнюю, если from_end=True) подходящую запись.
    """
    kwargs.setdefault("limit", 1)
    res = self.search(**kwargs)
    return res[0] if res else None

def save_records_to_json(self, filename: str) -> None:
    """Сохраняет список объектов RequestRecord в JSON файл."""
    with open(filename, "w", encoding="utf-8") as f:
        json.dump([asdict(r) for r in self._records], f, ensure_ascii=False, indent=2)

async def _lowered_headers(headers: Dict[str, str]) -> Dict[str, str]:
# playwright уже возвр. нормальный dict, нормализуем ключи для поиска
return {str(k).lower(): v for k, v in headers.items()}

def _headers_contains(
haystack: Dict[str, str],
needle: Dict[str, str],
) -> bool:
"""
Проверяем, что haystack содержит ВСЕ ключи/значения needle (ключи — lowercase).
Значения сверяем как подстроки (удобнее при длинных токенах/куках).
"""
h = {k.lower(): str(v) for k, v in haystack.items()}
for k, v in needle.items():
k_l = k.lower()
if k_l not in h:
return False
if str(v) not in h[k_l]:
return False
return True

def _json_matches(root: _JSON, expected: Dict[str, Any]) -> bool:
"""
Проверяем, что JSON содержит заданные значения по дот-путям.
Поддержка путей вида: "a.b[0].c"
"""
for path, want in expected.items():
got = _json_get(root, path)
if got != want:
return False
return True

_index_re = re.compile(r"(.+)[(\d+)]$")

def _json_get(root: _JSON, path: str) -> Any:
cur: Any = root
parts = [p for p in path.split(".") if p]
for p in parts:
m = _index_re.match(p)
if m:
name, idx_s = m.group(1), m.group(2)
idx = int(idx_s)
if not isinstance(cur, dict) or name not in cur:
return None
cur = cur[name]
if not isinstance(cur, list) or idx >= len(cur):
return None
cur = cur[idx]
continue

    # обычное поле
    if isinstance(cur, dict):
        cur = cur.get(p)
    else:
        return None
return cur

self.traffic = RequestHistory(
capture_bodies=True,
max_body_bytes=1_000_000,
decode="utf-8",
)
await self.traffic.attach(self.context)