Source code for simvx.core.assets.source

"""Source protocol: abstract storage backend that yields bytes for a URI.

Sources do *not* parse: they just resolve a URI to bytes (and optional
versioning metadata). Loaders parse those bytes into typed assets.

URI schemes supported by built-in Sources:

* ``pkg://<package>/<path>``: Python package resource (importlib.resources).
* ``file:///<absolute path>``: filesystem.
* Bare relative path: also routed to ``FileSource`` for ergonomics.
* ``http://...`` / ``https://...``: network (HttpSource).
* ``mem://<key>``: in-memory test source.

User code rarely touches Sources directly; the
:class:`~simvx.core.assets.server.AssetServer` looks up the right Source
from a URI's scheme.
"""

from __future__ import annotations

from collections.abc import Iterable
from typing import Protocol, runtime_checkable




@runtime_checkable
class Source(Protocol):
    """Storage backend mapped to a URI scheme.

    Implementations should be cheap to construct and stateless beyond
    configuration; the ``AssetServer`` shares one instance per scheme.
    """

    scheme: str  # e.g. "pkg", "file", "https"



    def read_bytes(self, uri: str) -> bytes:
        """Resolve *uri* and return the contents as raw bytes.

        Should raise :class:`FileNotFoundError`, :class:`OSError`, or a
        source-specific exception on failure. Network sources may raise
        :class:`ConnectionError` or :class:`TimeoutError`.
        """
        ...




    def version(self, uri: str) -> str | None:
        """Return a version hint (etag/mtime/None) for cache invalidation.

        Implementations that don't support versioning return ``None``.
        ``AssetServer`` uses the value as part of the cache key.
        """
        ...




    def list(self, uri: str) -> Iterable[str]:
        """List child URIs under *uri* (folder enumeration).

        Sources that cannot list (e.g. HTTP without directory indexing)
        should raise :class:`NotImplementedError`.
        """
        ...