added bugis.server package

server/example/test_server.py

@@ -0,0 +1,12 @@
+
+from bugis.core import BugisApp
+from pathlib import PurePath
+from bugis.server.server import static_resources
+import os
+
+root = os.getenv('STATIC_ROOT') or '.'
+app = BugisApp()
+
+static_resources(app, '/view', root)
+
+

server/pyproject.toml

@@ -0,0 +1,67 @@
+[build-system]
+requires = ["setuptools>=61.0", "setuptools-scm>=8"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "bugis_server"
+dynamic = ["version"]
+authors = [
+  { name="Walter Oggioni", email="oggioni.walter@gmail.com" },
+]
+description = "Static file renderer"
+readme = "README.md"
+requires-python = ">=3.10"
+classifiers = [
+        'Development Status :: 3 - Alpha',
+        'Topic :: Utilities',
+        'License :: OSI Approved :: MIT License',
+        'Intended Audience :: System Administrators',
+        'Intended Audience :: Developers',
+        'Environment :: Console',
+        'License :: OSI Approved :: MIT License',
+        'Programming Language :: Python :: 3',
+]
+dependencies = [
+    "bugis_core",
+    "Markdown",
+    "Pygments",
+    "watchdog",
+    "pwo",
+    "PyYAML",
+    "pygraphviz",
+    "aiofiles",
+    "httpx[http2]"
+]
+
+[project.optional-dependencies]
+dev = [
+    "build", "granian", "mypy", "ipdb", "twine"
+]
+
+run = [
+    "granian"
+]
+
+[tool.setuptools.package-data]
+bugis = ['static/*', 'default-conf/*']
+
+[project.urls]
+"Homepage" = "https://github.com/woggioni/bugis"
+"Bug Tracker" = "https://github.com/woggioni/bugis/issues"
+
+[tool.mypy]
+python_version = "3.12"
+disallow_untyped_defs = true
+show_error_codes = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unused_ignores = true
+exclude = ["scripts", "docs", "test"]
+strict = true
+
+[tool.setuptools_scm]
+root='..'
+version_file = "src/bugis/server/_version.py"
+
+[project.scripts]
+bugis = "bugis.server.cli:main"

server/src/bugis/server/cache.py

@@ -0,0 +1,16 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+
+class Cache(ABC):
+
+    @abstractmethod
+    def get(self, key: bytes) -> Optional[bytes]:
+        pass
+
+    @abstractmethod
+    def put(self, key: bytes, value: bytes):
+        pass
+
+    def __contains__(self, key: bytes):
+        return self.get(key) is not None

server/src/bugis/server/cli.py

@@ -0,0 +1,3 @@
+
+def main():
+    pass

server/src/bugis/server/server.py

@@ -0,0 +1,129 @@
+from base64 import b64encode
+from hashlib import md5
+from mimetypes import guess_type
+from pathlib import Path, PurePath
+from typing import TYPE_CHECKING, Optional, Callable, Awaitable, AsyncGenerator, Any, Unpack
+
+from aiofiles.os import listdir
+from aiofiles.ospath import isdir, isfile
+from bugis.core import HttpContext, BugisApp
+from pwo import Maybe
+
+if TYPE_CHECKING:
+    from _typeshed import StrOrBytesPath
+
+
+def parse_etag(etag: str) -> Optional[str]:
+    def skip_weak_marker(s):
+        if s.startswith('W/'):
+            return s[2:]
+        else:
+            return s
+
+    return (
+        Maybe.of_nullable(etag)
+        .map(skip_weak_marker)
+        .or_else(None)
+    )
+
+
+def static_resources(app: BugisApp,
+                     path: str,
+                     root: 'StrOrBytesPath',
+                     favicon: PurePath = None,
+                     file_filter: Callable[[PurePath], Awaitable[bool]] = None
+                     ):
+    async def no_filter(_: PurePath):
+        return True
+
+    if file_filter is None:
+        file_filter = no_filter
+
+    def compute_etag(resource: Path) -> str:
+        md = md5()
+        print(resource)
+        with resource.open('rb') as file:
+            while True:
+                chunk = file.read(0x10000)
+                if len(chunk):
+                    md.update(chunk)
+                else:
+                    break
+        return b64encode(md.digest()).decode()
+
+    if isinstance(root, str):
+        folder = Path(root)
+    else:
+        folder = root
+
+    prefix = PurePath(path)
+
+    async def handler(context: HttpContext, *_: Unpack[Any]) -> None:
+        requested = (PurePath(context.path)).relative_to(prefix)
+        resource = folder / requested
+        if not resource.is_relative_to(folder) or not resource.exists():
+            return await context.send_empty(404)
+
+        if await isfile(resource):
+
+            e_tag_header = ((Maybe.of_nullable(context.headers.get('if-none-match'))
+                             .filter(lambda it: len(it) > 0)
+                             .map(lambda it: it[-1])
+                             .map(parse_etag))
+                            .or_none())
+            current_etag = compute_etag(resource)
+
+            if e_tag_header == current_etag:
+                return await context.send_empty(304)
+            else:
+                mime_type = (Maybe.of(guess_type(resource.name))
+                             .map(lambda it: it[0])
+                             .or_else('application/octet-stream'))
+                return await context.send_file(200, resource, {
+                    'content-type': mime_type or 'application/octet-stream',
+                    'etag': 'W/' + current_etag,
+                    'cache-control': 'no-cache',
+                })
+        elif isdir(resource):
+            headers = {
+                'content-type': 'text/html'
+            }
+            await context.send_str(200, await directory_listing(prefix, requested, resource, favicon, file_filter),
+                                   headers)
+
+    return app.GET(path, True)(handler)
+
+
+async def directory_listing(prefix: PurePath,
+                            path_info: PurePath,
+                            path: PurePath,
+                            favicon: PurePath,
+                            file_filter: Callable[[PurePath], Awaitable[bool]]) -> str:
+    title = "Directory listing for %s" % path_info
+    result = "<!DOCTYPE html><html><head>"
+    if favicon:
+        result += f'<link rel="icon" type="image/x-icon" href="{favicon}">'
+    result += "<meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\">"
+    result += "<title>" + title + "</title></head>"
+    result += "<body><h1>" + title + "</h1><hr>"
+    result += "<ul>"
+    if path_info != '/':
+        result += "<li><a href=\"../\"/>../</li>"
+
+    async def ls(entry_filter: Callable[[PurePath], Awaitable[bool]]) -> AsyncGenerator[str, Any]:
+        async def result():
+            for entry in sorted(await listdir(path)):
+                if await entry_filter(path / entry):
+                    yield entry
+
+        return result()
+
+    async for entry in await ls(isdir):
+        result += '<li><a href="' + entry + '/' + '"/>' + entry + '/' + '</li>'
+
+    async def composite_file_filter(entry: PurePath) -> bool:
+        return await isfile(entry) and await file_filter(entry)
+
+    async for entry in await ls(composite_file_filter):
+        result += '<li><a href="' + entry + '"/>' + entry + '</li>'
+    return result