Add diypedals.fm sub-site

Makefile

@@ -34,26 +34,33 @@ install-python: venv
 ##
 
 build: # Build assets
-	make pybuild
+	make pybuild-www
+	make pybuild-diypedals
 	make messagesc
 
-pybuild:
+pybuild-www:
 	${bin}python -m tools.pybuild ./server/web/styles/styles.css --outdir ./server/web/static/css ${ARGS}
 
+pybuild-diypedals:
+	${bin}python -m tools.pybuild ./server/diypedals/web/styles/styles.css --outdir ./server/diypedals/web/static/css ${ARGS}
+
 ##
 ## ----------------
 ## Serving
 ## ----------------
 ##
 
 start: # Run servers
-	make -j start-uvicorn start-assets
+	make -j start-uvicorn start-assets-www start-assets-diypedals
 
 start-uvicorn:
 	PYTHONUNBUFFERED=1 ${bin}python -m server.main 2>&1 | ${bin}python -m tools.colorprefix blue [server]
 
-start-assets:
-	PYTHONUNBUFFERED=1 make pybuild ARGS="--watch" 2>&1 | ${bin}python -m tools.colorprefix yellow [assets]
+start-assets-www:
+	PYTHONUNBUFFERED=1 make pybuild-www ARGS="--watch" 2>&1 | ${bin}python -m tools.colorprefix yellow [assets-www]
+
+start-assets-diypedals:
+	PYTHONUNBUFFERED=1 make pybuild-diypedals ARGS="--watch" 2>&1 | ${bin}python -m tools.colorprefix red [assets-diypedals]
 
 ##
 ## ----------------

server/diypedals/.gitignore

@@ -0,0 +1,2 @@
+cache/
+web/static/img/build_reports/

server/diypedals/di.py

@@ -0,0 +1,48 @@
+from typing import TypeVar
+
+import httpx
+
+from . import settings
+from .domain.repositories import BuildReportRepository
+from .infrastructure.cache import DiskCache
+from .infrastructure.di import Container
+from .infrastructure.webdav import BuildReportClient
+
+T = TypeVar("T")
+
+
+def _configure(container: Container) -> None:
+    from .infrastructure.database import InMemoryDatabase
+    from .infrastructure.repositories import InMemoryBuildReportRepository
+    from .web.reload import HotReload
+    from .web.templating import Templates
+
+    db = InMemoryDatabase()
+    container.register(InMemoryDatabase, instance=db)
+
+    hotreload = HotReload()
+    container.register(HotReload, instance=hotreload)
+
+    container.register(Templates, instance=Templates(hotreload))
+
+    container.register(
+        BuildReportRepository, instance=InMemoryBuildReportRepository(db)
+    )
+    container.register(
+        BuildReportClient,
+        instance=BuildReportClient(
+            username=settings.WEBDAV_USERNAME,
+            password=str(settings.WEBDAV_PASSWORD),
+            cache=DiskCache(directory=settings.BUILD_REPORTS_CACHE_DIR),
+        ),
+    )
+
+
+def create_container() -> Container:
+    return Container(_configure)
+
+
+_container = create_container()
+
+bootstrap = _container.bootstrap
+resolve = _container.resolve

server/diypedals/domain/entities.py

@@ -0,0 +1,37 @@
+import datetime as dt
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class Kit:
+    name: str
+    seller: str
+    url: str
+
+
+@dataclass(frozen=True)
+class Pcb:
+    name: str
+    seller: str
+    url: str
+
+
+@dataclass(frozen=True)
+class Photo:
+    filename: str
+    content_type: str
+    alt: str
+
+
+@dataclass(frozen=True)
+class BuildReport:
+    title: str
+    slug: str
+    description: str
+    categories: list[str]
+    build_date: dt.date
+    status: str
+    thumbnail: Photo
+    photos: list[Photo] = field(default_factory=list)
+    kit: Kit | None = None
+    pcb: Pcb | None = None

server/diypedals/domain/repositories.py

@@ -0,0 +1,18 @@
+from .entities import BuildReport
+
+
+class BuildReportRepository:
+    async def save(self, build_report: BuildReport) -> None:
+        raise NotImplementedError  # pragma: no cover
+
+    async def find_all(self, *, category: str | None = None) -> list[BuildReport]:
+        raise NotImplementedError  # pragma: no cover
+
+    async def find_one(self, *, slug: str) -> BuildReport | None:
+        raise NotImplementedError  # pragma: no cover
+
+    async def get_unique_categories(self) -> dict[str, int]:
+        raise NotImplementedError  # pragma: no cover
+
+    async def find_all_with_category(self, category: str) -> list[BuildReport]:
+        raise NotImplementedError  # pragma: no cover

server/diypedals/infrastructure/cache.py

@@ -0,0 +1,45 @@
+import json
+import mimetypes
+from pathlib import Path
+from typing import Any, Callable
+
+
+class DiskCache:
+    def __init__(self, directory: Path) -> None:
+        self._dir = directory
+        self._dir.mkdir(exist_ok=True)
+        self._file = self._dir / "cache.json"
+        self._data: dict = {}
+
+        try:
+            self._data = json.loads(self._file.read_text())
+        except FileNotFoundError:
+            self._file.write_text("{}")
+
+    def _write_cache(self) -> None:
+        self._file.write_text(json.dumps(self._data))
+
+    async def get(self, key: str, fetch_func: Callable) -> Any:
+        try:
+            return self._data[key]
+        except KeyError:
+            value = await fetch_func()
+            self._data[key] = value
+            self._write_cache()
+            return value
+
+    async def get_file(
+        self, path: Path | str, fetch_func: Callable
+    ) -> tuple[bytes, str]:
+        path = Path(path)
+
+        try:
+            content = path.read_bytes()
+        except FileNotFoundError:
+            content, media_type = await fetch_func()
+            path.write_bytes(content)
+        else:
+            media_type = mimetypes.guess_type(path.name)
+            assert media_type is not None
+
+        return content, media_type

server/diypedals/infrastructure/content.py

@@ -0,0 +1,13 @@
+from .. import settings
+from ..di import resolve
+from ..domain.repositories import BuildReportRepository
+from .cache import DiskCache
+from .webdav import BuildReportClient
+
+
+async def load_build_reports() -> None:
+    repository = resolve(BuildReportRepository)
+    client = resolve(BuildReportClient)
+
+    async for build_report in client.fetch_all():
+        await repository.save(build_report)

server/diypedals/infrastructure/database.py

@@ -0,0 +1,68 @@
+from contextlib import contextmanager
+from typing import Iterator
+
+from .. import settings
+from ..domain.entities import BuildReport
+from .content import load_build_reports
+
+
+class _Data:
+    def __init__(self) -> None:
+        self.build_reports: dict[str, BuildReport] = {}
+
+
+class InMemoryDatabase:
+    def __init__(self) -> None:
+        self._data = _Data()
+
+    async def connect(self) -> None:
+        await self._load()
+
+    async def reload(self) -> None:
+        await self._load()  # pragma: no cover
+
+    @contextmanager
+    def isolated(self) -> Iterator[None]:
+        previous_data = self._data
+        self._data = _Data()
+        try:
+            yield
+        finally:
+            self._data = previous_data
+
+    async def _load(self) -> None:
+        self._data = _Data()
+
+        await load_build_reports()
+
+    def find_all_build_reports(
+        self, *, category: str | None = None
+    ) -> list[BuildReport]:
+        queryset = (build_report for build_report in self._data.build_reports.values())
+
+        if category is not None:
+            queryset = (
+                build_report
+                for build_report in queryset
+                if category in build_report.categories
+            )
+
+        return sorted(
+            queryset, key=lambda build_report: build_report.build_date, reverse=True
+        )
+
+    def find_one_build_report(self, slug: str) -> BuildReport | None:
+        return self._data.build_reports.get(slug)
+
+    def insert_build_report(self, build_report: BuildReport) -> None:
+        self._data.build_reports[build_report.slug] = build_report
+
+    def find_unique_build_report_categories(self) -> dict[str, int]:
+        categories_with_count: dict[str, int] = {}
+
+        for build_report in self._data.build_reports.values():
+            for category in build_report.categories:
+                categories_with_count.setdefault(category, 0)
+                categories_with_count[category] += 1
+
+        return categories_with_count

server/diypedals/infrastructure/di.py

@@ -0,0 +1,26 @@
+from typing import Callable, Type, TypeVar
+
+import punq
+
+T = TypeVar("T")
+
+
+class Container:
+    def __init__(self, configure: Callable[["Container"], None]) -> None:
+        self._impl = punq.Container()
+        self._configure = configure
+        self._is_configured = False
+
+    def register(self, type_: Type[T], *, instance: T) -> None:
+        self._impl.register(type_, instance=instance)
+
+    def bootstrap(self) -> None:
+        if self._is_configured:
+            return  # pragma: no cover
+
+        self._configure(self)
+        self._is_configured = True
+
+    def resolve(self, type_: Type[T]) -> T:
+        assert self._is_configured, "DI not configured: call bootstrap()"
+        return self._impl.resolve(type_)

server/diypedals/infrastructure/repositories.py

@@ -0,0 +1,20 @@
+from ..domain.entities import BuildReport
+from ..domain.repositories import BuildReportRepository
+from .database import InMemoryDatabase
+
+
+class InMemoryBuildReportRepository(BuildReportRepository):
+    def __init__(self, db: InMemoryDatabase) -> None:
+        self._db = db
+
+    async def save(self, build_report: BuildReport) -> None:
+        self._db.insert_build_report(build_report)
+
+    async def find_all(self, *, category: str | None = None) -> list[BuildReport]:
+        return self._db.find_all_build_reports(category=category)
+
+    async def find_one(self, *, slug: str) -> BuildReport | None:
+        return self._db.find_one_build_report(slug)
+
+    async def get_unique_categories(self) -> dict[str, int]:
+        return self._db.find_unique_build_report_categories()

server/diypedals/infrastructure/urls.py

@@ -0,0 +1,27 @@
+from typing import Any
+
+import httpx
+
+from .. import settings
+from ..domain.entities import BuildReport
+
+
+def to_production_url(url: str) -> str:
+    urlobj = httpx.URL(url)
+
+    scheme, host, port = (
+        ("http", f"localhost", settings.PORT)
+        if settings.DEBUG
+        else ("http" if settings.TESTING else "https", f"florimond.dev", None)
+    )
+
+    path = "/diypedals" + urlobj.path
+
+    return str(urlobj.copy_with(scheme=scheme, host=host, port=port, path=path))
+
+
+def get_absolute_path(obj: Any) -> str:
+    if isinstance(obj, BuildReport):
+        return f"/build-reports/{obj.slug}"
+
+    raise ValueError(f"No absolute URL defined for {obj!r}")