Source code for ewoksserver.app.backends.binary_backend

import base64
import logging
import mimetypes
import re
from pathlib import Path
from typing import Iterator

ResourceIdentifierType = str
ResourceUrlType = Path
ResourceContentType = dict

_logger = logging.getLogger(__name__)




def root_url(root_url: str | Path | None, category: str) -> ResourceUrlType:
    if not root_url:
        root_url = Path(".")
    elif isinstance(root_url, str):
        root_url = Path(root_url)
    return root_url / category





def resource_identifiers(root: ResourceUrlType) -> Iterator[ResourceIdentifierType]:
    for url in _resource_urls(root):
        yield _url_to_identifier(url)





def resources(root: ResourceUrlType) -> Iterator[ResourceContentType]:
    for url in _resource_urls(root):
        yield _load_url(url)





def resource_exists(root: ResourceUrlType, identifier: ResourceIdentifierType) -> bool:
    return _identifier_to_url(root, identifier).exists()



def _resource_urls(root: ResourceUrlType) -> Iterator[ResourceUrlType]:
    for url in root.iterdir():
        if _is_resource(url):
            yield url


def _is_resource(url: ResourceUrlType) -> bool:
    return (
        url.is_file() and not url.name.startswith(".") and not url.name.endswith(".py")
    )




def save_resource(
    root: ResourceUrlType,
    identifier: ResourceIdentifierType,
    resource: ResourceContentType,
):
    url = _identifier_to_url(root, identifier)
    _save_url(url, resource)





def load_resource(
    root: ResourceUrlType, identifier: ResourceIdentifierType
) -> ResourceContentType:
    url = _identifier_to_url(root, identifier)
    return _load_url(url)





def delete_resource(root: ResourceUrlType, identifier: ResourceIdentifierType) -> None:
    url = _identifier_to_url(root, identifier)
    _delete_url(url)



def _delete_url(url: ResourceUrlType) -> ResourceContentType:
    _logger.debug("Delete file '%s'", url)
    url.unlink()


def _identifier_to_url(
    root: ResourceUrlType, identifier: ResourceIdentifierType
) -> ResourceUrlType:
    url = (root / identifier).resolve()
    if not url.is_relative_to(root.resolve()):
        raise ValueError(f"Invalid resource identifier: {identifier!r}")
    return url


def _url_to_identifier(url: ResourceUrlType) -> ResourceIdentifierType:
    return url.name


def _load_url(url: ResourceUrlType) -> ResourceContentType:
    mimetype, encoding = mimetypes.guess_type(url, strict=True)
    if mimetype is None:
        raise ValueError(f"Cannot derive mime type from '{url}'")

    try:
        with open(url, "rb") as f:
            data = f.read()
    except FileNotFoundError:
        _logger.error(f"'{url}' not found")
        raise

    if not encoding:
        encoding = "base64"
        data = base64.b64encode(data).decode()

    return {"data_url": f"data:{mimetype};{encoding},{data}"}


def _save_url(url: ResourceUrlType, resource: ResourceContentType) -> None:
    _logger.debug("Save file '%s'", url)
    url.parent.mkdir(parents=True, exist_ok=True)
    data = _decode_data_url(resource["data_url"])
    with open(url, "wb") as f:
        f.write(data)


# Matches the "data:<mimetype>;base64," prefix produced by `_load_url` above.
# Exposed so API models (e.g. EwoksIcon) can validate this at the request boundary.
DATA_URL_PREFIX = re.compile(r"^data:[^;,]*;base64,", re.IGNORECASE)


def _decode_data_url(data_url: str) -> bytes:
    # Decode directly instead of using `urllib.request.urlopen`, which would
    # also follow "file:", "http:" and other schemes for client-provided input.
    match = DATA_URL_PREFIX.match(data_url)
    if not match:
        raise ValueError("Only base64-encoded 'data:' URLs are supported")
    return base64.b64decode(data_url[match.end() :])