Adds support for the Gotenberg webhooks

.pre-commit-config.yaml

@@ -39,10 +39,10 @@ repos:
         exclude: "(^Pipfile\\.lock$)"
   # Python hooks
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.1.6'
+    rev: 'v0.1.11'
     hooks:
       - id: ruff
-  - repo: https://github.com/psf/black
-    rev: 23.11.0
+  - repo: https://github.com/psf/black-pre-commit-mirror
+    rev: 23.12.1
     hooks:
       - id: black

CHANGELOG.md

@@ -11,6 +11,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Documentation site built with Github Pages and Material for MkDocs
 - New method `.run_with_retry` for routes, which allows the route to be rerun as configured, with progressive backoff if the server returns a server error
+- Support for Gotenberg [Webhooks](https://gotenberg.dev/docs/webhook)
+
+### Deprecated
+
+- Support for Gotenberg 7.x. This will likely be the last release to support 7.x, as the options for PDF/A have been changed
 
 ## [0.4.1] - 2023-12-11
 

src/gotenberg_client/_client.py

@@ -15,6 +15,7 @@
 from gotenberg_client._health import HealthCheckApi
 from gotenberg_client._merge import MergeApi
 from gotenberg_client._typing_compat import Self
+from gotenberg_client.options import HttpMethods
 
 
 class GotenbergClient:
@@ -44,12 +45,44 @@ def __init__(
         self.merge = MergeApi(self._client)
         self.health = HealthCheckApi(self._client)
 
-    def add_headers(self, header: Dict[str, str]) -> None:  # pragma: no cover
+    def add_headers(self, header: Dict[str, str]) -> None:
         """
         Updates the httpx Client headers with the given values
         """
         self._client.headers.update(header)
 
+    def add_webhook_url(self, url: str) -> None:
+        """
+        Adds the webhook URL to the headers
+        """
+        self.add_headers({"Gotenberg-Webhook-Url": url})
+
+    def add_error_webhook_url(self, url: str) -> None:
+        """
+        Adds the webhook error URL to the headers
+        """
+        self.add_headers({"Gotenberg-Webhook-Error-Url": url})
+
+    def set_webhook_http_method(self, method: HttpMethods = "PUT") -> None:
+        """
+        Sets the HTTP method Gotenberg will use to call the hooks
+        """
+        self.add_headers({"Gotenberg-Webhook-Method": method})
+
+    def set_error_webhook_http_method(self, method: HttpMethods = "PUT") -> None:
+        """
+        Sets the HTTP method Gotenberg will use to call the hooks
+        """
+        self.add_headers({"Gotenberg-Webhook-Error-Method": method})
+
+    def set_webhook_extra_headers(self, extra_headers: Dict[str, str]) -> None:
+        """
+        Sets the HTTP method Gotenberg will use to call the hooks
+        """
+        from json import dumps
+
+        self.add_headers({"Gotenberg-Webhook-Extra-Http-Headers": dumps(extra_headers)})
+
     def __enter__(self) -> Self:
         return self
 

src/gotenberg_client/options.py

@@ -5,6 +5,7 @@
 import enum
 from typing import Dict
 from typing import Final
+from typing import Literal
 from typing import Optional
 from typing import Union
 
@@ -102,3 +103,6 @@ def to_form(self) -> Dict[str, str]:
             return {"emulatedMediaType": "screen"}
         else:  # pragma: no cover
             raise NotImplementedError(self.value)
+
+
+HttpMethods = Literal["POST", "PATCH", "PUT"]

tests/test_misc_stuff.py

@@ -1,10 +1,13 @@
 import shutil
 import tempfile
 import uuid
+from json import dumps
+from json import loads
 from pathlib import Path
 
 import pytest
 from httpx import HTTPStatusError
+from httpx import Request
 from httpx import codes
 from pytest_httpx import HTTPXMock
 
@@ -105,3 +108,71 @@ def test_not_a_server_error(self, client: GotenbergClient, httpx_mock: HTTPXMock
             with pytest.raises(HTTPStatusError) as exc_info:
                 _ = route.index(test_file).run_with_retry(initial_retry_wait=0.1, retry_scale=0.1)
             assert exc_info.value.response.status_code == codes.NOT_FOUND
+
+
+class TestWebhookHeaders:
+    def test_webhook_basic_headers(self, client: GotenbergClient, httpx_mock: HTTPXMock):
+        httpx_mock.add_response(method="POST", status_code=codes.OK)
+
+        client.add_webhook_url("http://myapi:3000/on-success")
+        client.add_error_webhook_url("http://myapi:3000/on-error")
+
+        test_file = SAMPLE_DIR / "basic.html"
+        with client.chromium.html_to_pdf() as route:
+            _ = route.index(test_file).run_with_retry()
+
+        requests = httpx_mock.get_requests()
+
+        assert len(requests) == 1
+
+        request: Request = requests[0]
+
+        assert "Gotenberg-Webhook-Url" in request.headers
+        assert request.headers["Gotenberg-Webhook-Url"] == "http://myapi:3000/on-success"
+        assert "Gotenberg-Webhook-Error-Url" in request.headers
+        assert request.headers["Gotenberg-Webhook-Error-Url"] == "http://myapi:3000/on-error"
+
+    def test_webhook_http_methods(self, client: GotenbergClient, httpx_mock: HTTPXMock):
+        httpx_mock.add_response(method="POST", status_code=codes.OK)
+
+        client.add_webhook_url("http://myapi:3000/on-success")
+        client.set_webhook_http_method("POST")
+        client.add_error_webhook_url("http://myapi:3000/on-error")
+        client.set_error_webhook_http_method("GET")
+
+        test_file = SAMPLE_DIR / "basic.html"
+        with client.chromium.html_to_pdf() as route:
+            _ = route.index(test_file).run_with_retry()
+
+        requests = httpx_mock.get_requests()
+
+        assert len(requests) == 1
+
+        request: Request = requests[0]
+
+        assert "Gotenberg-Webhook-Method" in request.headers
+        assert request.headers["Gotenberg-Webhook-Method"] == "POST"
+        assert "Gotenberg-Webhook-Error-Method" in request.headers
+        assert request.headers["Gotenberg-Webhook-Error-Method"] == "GET"
+
+    def test_webhook_extra_headers(self, client: GotenbergClient, httpx_mock: HTTPXMock):
+        httpx_mock.add_response(method="POST", status_code=codes.OK)
+
+        headers = {"Token": "mytokenvalue"}
+        headers_str = dumps(headers)
+
+        client.set_webhook_extra_headers(headers)
+
+        test_file = SAMPLE_DIR / "basic.html"
+        with client.chromium.html_to_pdf() as route:
+            _ = route.index(test_file).run_with_retry()
+
+        requests = httpx_mock.get_requests()
+
+        assert len(requests) == 1
+
+        request: Request = requests[0]
+
+        assert "Gotenberg-Webhook-Extra-Http-Headers" in request.headers
+        assert request.headers["Gotenberg-Webhook-Extra-Http-Headers"] == headers_str
+        assert loads(request.headers["Gotenberg-Webhook-Extra-Http-Headers"]) == headers