feat!: replace docs command by mkdocs plugin wakemedoc

pyproject.toml

@@ -15,8 +15,12 @@ classifiers = [
 readme = "README.md"
 packages = [
     {include = "wakemebot", from  = "src"},
+    {include = "wakemebot", from  = "src"},
 ]
 
+[tool.poetry.plugins."mkdocs.plugins"]
+wakemedoc = "wakemebot.plugin:WakeMeDocPlugin"
+
 [tool.poetry.dependencies]
 python = ">=3.9,<3.11"
 httpx = ">=0.20.0,<0.22.0"
@@ -57,8 +61,6 @@ single_binary_application = """
 pyinstaller --onefile src/wakemebot/__main__.py --name wakemebot -s
 """
 
-
-
 [tool.black]
 line-length = 90
 target_version = ['py39']

setup.cfg

@@ -4,15 +4,15 @@ disallow_untyped_defs = True
 warn_redundant_casts = True
 strict_equality = True
 
-[mypy-anybadge]
-ignore_missing_imports = True
-
 [mypy-semver]
 ignore_missing_imports = True
 
 [mypy-typer]
 ignore_missing_imports = True
 
+[mypy-mkdocs.*]
+ignore_missing_imports = True
+
 [pydantic-mypy]
 init_forbid_extra = True
 init_typed = True

src/wakemebot/apt.py

@@ -1,7 +1,10 @@
+from collections import defaultdict
+from contextlib import contextmanager
 from dataclasses import dataclass, field
-from functools import lru_cache
+from itertools import product
+from pathlib import Path
 from textwrap import dedent
-from typing import Dict, List
+from typing import Callable, Dict, Generator, List, Set
 
 import httpx
 from debian.deb822 import Packages, Release
@@ -15,7 +18,8 @@ class RepositoryPackage:
     description: str
     homepage: str
     latest_version: str
-    versions: Dict[str, List[str]] = field(default_factory=dict)  # versions per arch
+    component: str
+    versions: Dict[str, Set[str]] = field(default_factory=lambda: defaultdict(set))
 
 
 @dataclass
@@ -27,30 +31,33 @@ class RepositoryComponent:
 
 @dataclass
 class Repository:
-    package_count: int
+    architectures: List[str]
     components: List[RepositoryComponent]
+    packages: Dict[str, RepositoryPackage]
+    package_count: int
 
 
-def _download_repository_release_file(client: httpx.Client, distribution: str) -> Release:
-    url = f"/dists/{distribution}/Release"
-    response = client.get(url)
-    response.raise_for_status()
-    return Release(response.text)
+@contextmanager
+def read_page_content(base_url: str) -> Generator[Callable[[str], str], None, None]:
+    with httpx.Client(base_url=base_url, follow_redirects=True) as client:
 
+        def _get(url: str) -> str:
+            content_path = Path("/tmp/wakemebot") / url[1:]
+            if content_path.exists():
+                return content_path.read_text()
+            else:
+                print(f"Downloading {url}...")
+                response = client.get(url)
+                response.raise_for_status()
+                content_path.parent.mkdir(exist_ok=True, parents=True)
+                content_path.write_text(response.text)
+                return response.text
 
-@lru_cache
-def _download_repository_packages_file(
-    client: httpx.Client, distribution: str, component: str, arch: str
-) -> bytes:
-    url = f"/dists/{distribution}/{component}/binary-{arch}/Packages"
-    print(f"Downloading {url}...")
-    with client.stream("GET", url) as response:
-        response.raise_for_status()
-        return response.read()
+        yield _get
 
 
 def _parse_repository_packages_file(
-    content: bytes, packages: Dict[str, RepositoryPackage]
+    component: str, content: str, packages: Dict[str, RepositoryPackage]
 ) -> None:
     """Extract package names and versions from a repo Packages file"""
     for src in Packages.iter_paragraphs(content, use_apt_pkg=False):
@@ -64,37 +71,36 @@ def _parse_repository_packages_file(
                 name=package_name,
                 summary=src["Description"].split("\n")[0],
                 latest_version=version or "unknown",
+                component=component,
             )
-        if architecture not in packages[package_name].versions:
-            packages[package_name].versions[architecture] = []
-        package_versions = packages[package_name].versions[architecture]
-        if version is not None and version not in package_versions:
-            package_versions.append(version)
-
-
-def _build_repository_component(
-    client: httpx.Client, component: str, architectures: List[str]
-) -> RepositoryComponent:
-    packages: Dict[str, RepositoryPackage] = {}
-    for architecture in architectures:
-        content = _download_repository_packages_file(
-            client, "stable", component, architecture
-        )
-        _parse_repository_packages_file(content, packages)
-    return RepositoryComponent(
-        name=component, packages=list(packages.values()), package_count=len(packages)
-    )
+        versions = packages[package_name].versions
+        if version is not None:
+            versions[version].add(architecture)
 
 
 def parse_repository(repository_url: str, distribution: str) -> Repository:
-    with httpx.Client(base_url=repository_url, follow_redirects=True) as client:
-        release = _download_repository_release_file(client, distribution)
+    with read_page_content(repository_url) as reader:
+        packages: Dict[str, RepositoryPackage] = {}
+        release = Release(reader(f"/dists/{distribution}/Release"))
         architectures = release["Architectures"].split(" ")
-        components = release["Components"].split(" ")
-        repository_components = [
-            _build_repository_component(client, c, architectures) for c in components
-        ]
+        component_names = release["Components"].split(" ")
+        for component, arch in product(component_names, architectures):
+            content = reader(f"/dists/stable/{component}/binary-{arch}/Packages")
+            _parse_repository_packages_file(component, content, packages)
+
+    components = []
+    for component in component_names:
+        component_packages = [p for p in packages.values() if p.component == component]
+        repository_component = RepositoryComponent(
+            name=component,
+            package_count=len(component_packages),
+            packages=component_packages,
+        )
+        components.append(repository_component)
+
     return Repository(
-        components=repository_components,
-        package_count=sum([c.package_count for c in repository_components]),
+        architectures=architectures,
+        components=components,
+        packages={k: v for k, v in sorted(packages.items())},
+        package_count=len(packages),
     )

src/wakemebot/badges.py

@@ -215,6 +215,7 @@ def generate(self) -> str:
         )
 
     def save(self, output_path: Path) -> None:
+        output_path.parent.mkdir(exist_ok=True, parents=True)
         output_path.write_text(self.generate())
 
 

src/wakemebot/cli.py

@@ -1,10 +1,9 @@
-import sys
 from pathlib import Path
 
 import typer
-from pydantic import BaseModel, Field, HttpUrl, ValidationError
+from pydantic import BaseModel, Field, HttpUrl
 
-from wakemebot import aptly, docs
+from wakemebot import aptly
 
 from . import __version__
 
@@ -22,34 +21,6 @@ def version() -> None:
     typer.secho(__version__)
 
 
-@app.command(name="docs", help="Update documentation")
-def update_documentation(
-    debian_repository: str = typer.Option(
-        "http://deb.wakemeops.com/wakemeops/ stable",
-        "--repository",
-        "-r",
-        envvar="OPS2DEB_REPOSITORY",
-        help='Format: "{debian_repository_url} {distribution_name}". '
-        'Example: "http://deb.wakemeops.com/ stable". '
-        "Packages already published in the repo won't be generated.",
-    ),
-) -> None:
-    try:
-        url, distribution = debian_repository.split(" ")
-    except ValueError:
-        print(
-            "The expected format for the --repository option is "
-            '"{repository_url} {distribution}"'
-        )
-        sys.exit(1)
-    try:
-        DebianRepository(url=url, distribution=distribution)
-    except ValidationError as e:
-        print(e)
-        sys.exit(1)
-    docs.update_documentation(url, distribution)
-
-
 @aptly_app.command(name="push", help="Push debian sources packages to aptly repository")
 def aptly_push(
     repository: str = typer.Argument(..., help="Aptly repository name"),