Source code for simvx.core.assets.sources.http

"""HTTP/HTTPS source: fetches bytes over the network.

Uses :mod:`urllib.request` from the standard library so the engine has no
required network dependency. The web backend overrides this with a
``pyodide.http.pyfetch`` shim (see ``simvx.web.assets``).

Listing is not supported: HTTP has no portable directory protocol. To
load a group of remote URIs, ship a manifest JSON listing the URIs and
call ``AssetServer.load_manifest(...)``.
"""

from __future__ import annotations

import urllib.request
from collections.abc import Iterable




class HttpSource:
    """Reads bytes via :func:`urllib.request.urlopen`.

    Captures the ``ETag`` response header (when present) and exposes it
    through :meth:`version` so the cache can revalidate without a full
    refetch on subsequent loads.
    """

    scheme = "https"  # Same impl serves both schemes; AssetServer registers each.

    def __init__(self, *, timeout: float = 10.0) -> None:
        self._etag_cache: dict[str, str] = {}
        self.timeout = timeout



    def read_bytes(self, uri: str) -> bytes:
        req = urllib.request.Request(uri)
        with urllib.request.urlopen(req, timeout=self.timeout) as resp:
            etag = resp.headers.get("ETag")
            if etag:
                self._etag_cache[uri] = etag
            return resp.read()




    def version(self, uri: str) -> str | None:
        cached = self._etag_cache.get(uri)
        if cached is not None:
            return f"etag:{cached}"
        try:
            req = urllib.request.Request(uri, method="HEAD")
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                etag = resp.headers.get("ETag")
                if etag:
                    self._etag_cache[uri] = etag
                    return f"etag:{etag}"
                last_mod = resp.headers.get("Last-Modified")
                if last_mod:
                    return f"last-modified:{last_mod}"
        except OSError:
            return None
        return None




    def list(self, uri: str) -> Iterable[str]:
        raise NotImplementedError(
            "HttpSource cannot list remote directories; ship a manifest JSON instead."
        )