Html resources

src/wolf/app/render/html.py

@@ -1,15 +1,44 @@
 import wrapt
 import structlog
+from pathlib import PurePosixPath
 from typing import Sequence
 from functools import partial
 from wolf.rendering.ui import UI
-from wolf.rendering.resources import Resource, NeededResources
 from wolf.app.response import Response, FileWrapperResponse
+from html_resources.resources import Resource
+from html_resources.needed import NeededResources
 
 
 logger = structlog.get_logger("wolf.app.render")
 
 
+class BoundResources(NeededResources):
+
+    def __init__(self, path: str | PurePosixPath, *args, **kwargs):
+        self.path = PurePosixPath(path)
+        super().__init__(*args, **kwargs)
+
+    def apply(self, body: str | bytes, base_uri: str = "") -> bytes:
+        if len(self.data) == 0:
+            return body
+
+        if isinstance(body, str):
+            body = body.encode()
+
+        top = b""
+        bottom = b""
+        for resource in self.unfold():
+            if resource.bottom:
+                bottom += resource.render(base_uri / self.path)
+            else:
+                top += resource.render(base_uri / self.path)
+        if top:
+            body = body.replace(b"</head>", top + b"</head>", 1)
+        if bottom:
+            body = body.replace(b"</body>", bottom + b"</body>", 1)
+        return body
+
+
 def html(func=None, *, resources: Sequence[Resource] | None = None):
     @wrapt.decorator
     def html_wrapper(
@@ -24,7 +53,7 @@ def html_wrapper(
             raise TypeError(f"Unable to render type: {type(content)}.")
 
         request = args[0]
-        needed_resources = request.get(NeededResources, default=None)
+        needed_resources = request.get(BoundResources, default=None)
         if needed_resources is None:
             logger.warning("No resource injection.")
         else:

src/wolf/app/services/resources.py

@@ -1,199 +1,38 @@
-import os
-import base64
-import hashlib
-import enum
-import importlib.resources
-from typing import Sequence
-from pathlib import PurePosixPath, Path
-from mimetypes import guess_type
-from autoroutes import Routes
+from pathlib import PurePosixPath
+from html_resources.store import Repository
 from wolf.app.nodes import Node
+from wolf.app.render.html import BoundResources
 from wolf.app.response import Response, FileWrapperResponse
-from wolf.rendering.resources import (
-    Resource, known_extensions, NeededResources)
 from wolf.app.pluggability import Installable
 
 
-class HashAlgorithm(enum.Enum):
-    sha256 = hashlib.sha256
-    sha384 = hashlib.sha384
-    sha512 = hashlib.sha512
+class ResourceManager(Installable, Node):
 
-
-def generate_hash(filepath: Path, algorithm: HashAlgorithm) -> str:
-    hashed = algorithm.value()
-    with filepath.open("rb") as f:
-        while True:
-            data = f.read(1024 * 32)
-            if not data:
-                break
-            hashed.update(data)
-    return hashed.digest()
-
-
-class BaseLibrary:
-    name: str
-    base_path: Path
-
-    def __init__(self, name: str, base_path: str | Path):
-        resource = Path(base_path)
-        if not resource.exists():
-            raise OSError(f"{resource} does not exist.")
-        if not resource.is_dir():
-            raise TypeError("Library base path must be a directory.")
-        if not base_path.is_absolute():
-            raise ValueError("Base path needs to be absolute.")
-        self.name = name
-        self.base_path = base_path
-
-    def __iter__(self):
-        pass
-
-
-class DiscoveryLibrary(BaseLibrary):
-    name: str
-    base_path: Path
-
-    def __init__(self, name: str, base_path: str | Path, restrict=("*",)):
-        super().__init__(name, base_path)
-        self.restrictions = restrict
-
-    def __iter__(self):
-        for matcher in self.restrictions:
-            for path in self.base_path.rglob(matcher):
-                yield self.name / path.relative_to(self.base_path), path
-
-
-class Library(DiscoveryLibrary):
-    _resources: set
-    _by_name: dict[str, Resource]
-
-    def __init__(self, name: str, base_path: str | Path, restrict=("*",)):
-        self._resources = set()
-        self._by_name = {}
-        super().__init__(name, base_path, restrict=restrict)
-
-    def bind(
-        self,
-        path: str | PurePosixPath,
-        *,
-        name: str | None = None,
-        bottom: bool = False,
-        dependencies: Sequence[Resource] | None = None,
-    ):
-        fullpath = self.base_path / path
-        if not fullpath.is_file():
-            raise TypeError(f"{path} is not a file.")
-
-        if not fullpath.suffix:
-            raise NameError("Filename needs an extension.")
-
-        ext = fullpath.suffix[1:]
-        cls = known_extensions.get(ext)
-        if not cls:
-            raise TypeError("Unknown extension.")
-
-        hash_base64 = base64.b64encode(
-            generate_hash(fullpath, HashAlgorithm.sha256),
-        ).decode("utf-8")
-        integrity = f"sha256-{hash_base64}"
-
-        if dependencies is not None:
-            dependencies = tuple(dependencies)
-        resource = cls(
-            f"/{self.name}/{path}",
-            bottom=bottom,
-            integrity=integrity,
-            dependencies=dependencies,
-        )
-        self._resources.add(resource)
-        if name:
-            self._by_name[name] = resource
-        return resource
-
-
-class StaticAccessor:
-    path: str
-    resources: Routes | None
-    libraries: dict[str, Library]
-
-    def __init__(self, path: str):
-        self.path = path
-        self.resources = Routes()
-        self.libraries = {}
-
-    def feed(self, library: Library):
-        for uri, full_path in iter(library):
-            stats = os.stat(full_path)
-            content_type, encoding = guess_type(full_path)
-            if not content_type:
-                content_type = "octet/steam"
-            elif (
-                content_type.startswith("text/")
-                or content_type == "application/javascript"
-            ):
-                content_type += "; charset=utf-8"
-            info = {
-                "filepath": full_path,
-                "size": stats.st_size,
-                "last_modified": stats.st_mtime,
-                "content_type": content_type,
-            }
-            self.resources.add(str("/" / PurePosixPath(uri)), **info)
-
-    def add_library(self, library: Library, override: bool = False):
-        if library.name in self.libraries and not override:
-            raise KeyError(f"Library {library.name!r} already exists.")
-        self.libraries[library.name] = library
-        self.feed(library)
-
-    def add_static(
-            self,
-            name: str,
-            base_path: str | Path,
-            restrict=("*",),
-            override: bool = False,
-    ) -> Library:
-        library = DiscoveryLibrary(name, base_path, restrict=restrict)
-        self.add_library(library, override=override)
-        return library
-
-    def add_package_static(
-        self, package_static: str, restrict=("*",), override: bool = False
-    ):
-        # package_static of form:  package_name:path
-        pkg, resource_name = package_static.split(":")
-        resource = Path(
-            importlib.resources.files(pkg) / resource_name
-        )
-        return self.add_static(
-            package_static, resource, restrict=restrict, override=override
-        )
-
-
-class ResourceManager(Installable, Node, StaticAccessor):
+    def __init__(self, repository: Repository, path: str | PurePosixPath):
+        self.repository = repository
+        self.path = PurePosixPath(path)
 
     def install(self, application):
         application.sinks[self.path] = self
-        application.services.register_value(
-            ResourceManager, self
-        )
         application.services.register_factory(
-            NeededResources, self.needed_resources)
+            BoundResources, self.needed_resources
+        )
 
-    def needed_resources(self) -> NeededResources:
-        return NeededResources(self.path)
+    def needed_resources(self):
+        return BoundResources(self.path)
 
     def resolve(self, environ):
-        match, _ = self.resources.match(environ["PATH_INFO"])
-        if not match:
+        info = self.repository.match(
+            PurePosixPath(environ["PATH_INFO"].lstrip('/'))
+        )
+        if not info:
             return Response(status=404)
 
         headers = {
-            "Content-Length": str(match["size"]),
-            "Content-Type": match["content_type"],
+            "Content-Length": str(info.size),
+            "Content-Type": info.content_type,
         }
         if environ["REQUEST_METHOD"] == "HEAD":
             return Response(200, headers=headers)
 
-        return FileWrapperResponse(match["filepath"], headers=headers)
+        return FileWrapperResponse(info.filepath, headers=headers)