Almost all routes, except metrics, implemented and tested

pyproject.toml

@@ -49,15 +49,14 @@ exclude = [
 
 [tool.hatch.envs.default]
 dependencies = [
-  "coverage[toml] >= 7.0",
-  "pytest >= 7.0",
+  "coverage[toml] >= 7.3",
+  "pytest >= 7.4",
   "pytest-sugar",
-  "pytest-cov",
-  "pytest-xdist",
   "pytest-httpx ~= 0.26; python_version >= '3.9'",
   "pytest-httpx ~= 0.22; python_version < '3.9'",
-  "pikepdf"
-
+  "pikepdf",
+  "python-magic",
+  "brotli",
 ]
 
 [tool.hatch.envs.default.scripts]
@@ -76,7 +75,8 @@ cov = [
   "cov-clear",
   "test-cov",
   "cov-report",
-  "cov-json"
+  "cov-json",
+  "cov-html"
 ]
 pip-list = "pip list"
 

src/gotenberg_client/base.py

@@ -0,0 +1,95 @@
+import logging
+from contextlib import ExitStack
+from importlib.util import find_spec
+from pathlib import Path
+from types import TracebackType
+from typing import Dict
+from typing import Optional
+from typing import Type
+
+from httpx import Client
+from httpx import Response
+from httpx._types import RequestFiles
+
+from gotenberg_client.pdf_format import PdfAFormatOptions
+
+logger = logging.getLogger(__name__)
+
+
+class BaseRoute:
+    def __init__(self, client: Client, api_route: str) -> None:
+        self._client = client
+        self._route = api_route
+        self._stack = ExitStack()
+        self._form_data: Dict[str, str] = {}
+        self._file_map: Dict[str, Path] = {}
+
+    def __enter__(self) -> "BaseRoute":
+        self.reset()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        self.reset()
+
+    def reset(self) -> None:
+        self._stack.close()
+        self._form_data.clear()
+        self._file_map.clear()
+
+    def run(self) -> Response:
+        resp = self._client.post(url=self._route, data=self._form_data, files=self.get_files())
+        resp.raise_for_status()
+        return resp
+
+    def get_files(self) -> RequestFiles:
+        files = {}
+        for filename in self._file_map:
+            file_path = self._file_map[filename]
+            # Gotenberg requires these to have the specific name
+            filepath_name = filename if filename in {"index.html", "header.html", "footer.html"} else file_path.name
+
+            mime_type = guess_mime_type(file_path)
+            if mime_type is not None:
+                files.update(
+                    {filepath_name: (filepath_name, self._stack.enter_context(file_path.open("rb")), mime_type)},
+                )
+            else:  # pragma: no cover
+                files.update({filepath_name: (filepath_name, self._stack.enter_context(file_path.open("rb")))})  # type: ignore
+        return files
+
+    def _add_file_map(self, filepath: Path, name: Optional[str] = None) -> None:
+        if name is None:
+            name = filepath.name
+        if name in self._file_map:  # pragma: no cover
+            logger.warning(f"{name} has already been provided, overwriting anyway")
+        self._file_map[name] = filepath
+
+    def pdf_format(self, pdf_format: PdfAFormatOptions) -> "BaseRoute":
+        self._form_data.update(pdf_format.to_form())
+        return self
+
+
+class BaseApi:
+    def __init__(self, client: Client) -> None:
+        self._client = client
+
+
+def guess_mime_type_stdlib(url: Path) -> Optional[str]:
+    import mimetypes
+
+    mime_type, _ = mimetypes.guess_type(url)
+    return mime_type
+
+
+def guess_mime_type_magic(url: Path) -> Optional[str]:
+    import magic  # type: ignore
+
+    return magic.from_file(url, mime=True)  # type: ignore
+
+
+guess_mime_type = guess_mime_type_magic if find_spec("magic") is not None else guess_mime_type_stdlib

src/gotenberg_client/client.py

@@ -1,12 +1,17 @@
 import logging
+from importlib.util import find_spec
 from types import TracebackType
 from typing import Dict
 from typing import Optional
 from typing import Type
 
 from httpx import Client
 
-from gotenberg_client.convert.chromium import ChromiumRoutes
+from gotenberg_client.convert.chromium import ChromiumApi
+from gotenberg_client.convert.libre_office import LibreOfficeApi
+from gotenberg_client.convert.pdfa import PdfAApi
+from gotenberg_client.health import HealthCheckApi
+from gotenberg_client.merge import MergeApi
 
 
 class GotenbergClient:
@@ -16,7 +21,6 @@ def __init__(
         gotenerg_url: str,
         timeout: float = 30.0,
         log_level: int = logging.ERROR,
-        compress: bool = False,
         http2: bool = True,
     ):
         # Configure the client
@@ -26,15 +30,18 @@ def __init__(
         logging.getLogger("httpx").setLevel(log_level)
         logging.getLogger("httpcore").setLevel(log_level)
 
-        # Only JSON responses supported
-        self._client.headers.update({"Accept": "application/json"})
-
-        if compress:
-            # TODO Brotli?
-            self._client.headers.update({"Accept-Encoding": "gzip"})
+        # TODO Brotli?
+        if find_spec("brotli") is not None:
+            self._client.headers.update({"Accept-Encoding": "gzip,deflate,br"})
+        else:
+            self._client.headers.update({"Accept-Encoding": "gzip,deflate"})
 
         # Add the resources
-        self.chromium = ChromiumRoutes(self._client)
+        self.chromium = ChromiumApi(self._client)
+        self.libre_office = LibreOfficeApi(self._client)
+        self.pdf_a = PdfAApi(self._client)
+        self.merge = MergeApi(self._client)
+        self.health = HealthCheckApi(self._client)
         # TODO
 
     def add_headers(self, header: Dict[str, str]) -> None:  # pragma: no cover