Merge pull request #353 from danidelvalle/348-feature-request-support…

…-auto-provisioning-checks

Support urls with query parameters, such as '?create=1'

README.rst

@@ -41,17 +41,20 @@ A simple python decorator for `healthchecks.io`_.
 Features
 --------
 
-* Just decorate your function with ``@healthcheck`` 🚀.
-* Support sending ``/start`` signals to measure job execution times ⏲️.
-* Automatic ``/failure`` signals when jobs produce exceptions 🔥.
-* Send diagnostics information 🌡️.
-* Support both SaaS and self-hosted endpoints 😊.
+The `healthchecks-decorator` library provides the following features:
+
+* 🚀 **Easy to use:** Simply decorate your function with `@healthcheck` to enable health checks.
+* ⏲️ **Execution time measurement:** Supports sending `/start` signals to measure job execution times.
+* 🔥 **Exception handling:** Automatically sends `/failure` signals when jobs produce exceptions.
+* 🤖 **Auto-provisioning:** Supports automatic provisioning of new health checks by adding `?create=1` to the ping URL.
+* 🌡️ **Diagnostics information:** Send diagnostics information to help diagnose issues.
+* 😊 **Flexible endpoint support:** Supports both SaaS and self-hosted endpoints.
 
 
 Requirements
 ------------
 
-* None - only pure python 🐍.
+* None - just pure python 🐍.
 
 
 Installation

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "healthchecks-decorator"
-version = "0.4.1"
+version = "0.5.0"
 description = "Healthchecks Decorator"
 authors = ["Daniel del Valle <delvalle.dani@gmail.com>"]
 license = "MIT"

src/healthchecks_decorator/decorator.py

@@ -6,6 +6,8 @@
 from functools import wraps
 from os import getenv
 from urllib.parse import urlencode
+from urllib.parse import urlparse
+from urllib.parse import urlunparse
 from urllib.request import urlopen
 
 WrappedFn = t.TypeVar("WrappedFn", bound=t.Callable[..., t.Any])
@@ -15,6 +17,7 @@
 
 
 ENV_VAR_PREFIX = "HEALTHCHECK"
+VALID_URL_SCHEMES = ("http", "https")
 
 
 @dataclass
@@ -25,6 +28,54 @@ class HealthcheckConfig:
     send_start: t.Optional[bool]
     send_diagnostics: t.Optional[bool]
 
+    def _build_url_with_path(self, path: str) -> str:
+        """Build a sub URL."""
+        parsed_url = urlparse(self.url)
+        sep = "" if parsed_url.path.endswith("/") else "/"  # type: ignore
+        new_path = parsed_url.path + sep + path  # type: ignore
+        new_url = urlunparse(
+            (  # type: ignore
+                parsed_url.scheme,
+                parsed_url.netloc,
+                new_path,
+                parsed_url.params,
+                parsed_url.query,
+                parsed_url.fragment,
+            )
+        )
+        return new_url  # type: ignore
+
+    @property
+    def start_url(self) -> str:
+        """Return the start URL."""
+        return self._build_url_with_path("start")
+
+    @property
+    def fail_url(self) -> str:
+        """Return the fail URL."""
+        return self._build_url_with_path("fail")
+
+    def __bool__(self) -> bool:
+        """Return True if the config is valid, False otherwise."""
+        if not self.url:
+            logging.warning("Missing URL")
+            return False
+
+        try:
+            parsed_url = urlparse(self.url)
+
+            if parsed_url.scheme not in VALID_URL_SCHEMES:
+                logging.warning(f"Invalid URL scheme for URL: {self.url}")
+                return False
+
+            if not parsed_url.netloc:
+                logging.warning(f"Invalid netloc for URL: {self.url}")
+                return False
+            return True
+        except AttributeError:
+            logging.warning(f"Invalid URL: {self.url}")
+            return False
+
     def __getattribute__(self, name: str) -> t.Any:
         """Overloaded to get info from environment variables or defaults when not defined."""
         candidate = super().__getattribute__(name)
@@ -58,19 +109,13 @@ def _http_request(
         timeout (int, optional): Connection timeout in seconds. Defaults to 10.
         data (bytes, optional): Optional diagnostic data. Defaults to None.
 
-    Raises:
-        ValueError: if the endpoint schema is not http or https
-
     Returns:
         bool: True if the request succeeded, False otherwise.
     """
     try:
-        if endpoint.lower().startswith("http"):
-            # Bandit will still complain, so S310 is omitted
-            urlopen(endpoint, data=data, timeout=timeout)  # noqa: S310
-            return True
-        else:
-            raise ValueError("Only http[s] schemes allowed.") from None
+        # Bandit will still complain, so S310 is omitted
+        urlopen(endpoint, data=data, timeout=timeout)  # noqa: S310
+        return True
     except OSError:
         return False
 
@@ -133,18 +178,15 @@ def healthcheck(
         url=url, send_diagnostics=send_diagnostics, send_start=send_start
     )
 
-    if config.url is None or not len(config.url):
-        log.warning("Disabling @healthcheck: 'url' argument is not provided")
+    if not config:
+        log.warning("Disabling @healthcheck: invalid config")
         return func
 
-    sep = "" if config.url.endswith("/") else "/"
-
     @wraps(func)
     def healthcheck_wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
         assert config.url is not None  # noqa: S101
         if config.send_start:
-            url_with_start = f"{config.url}{sep}start"
-            _http_request(url_with_start)
+            _http_request(config.start_url)
 
         try:
             wrapped_result = func(*args, **kwargs)
@@ -156,8 +198,7 @@ def healthcheck_wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
             )
             return wrapped_result
         except Exception as e:
-            url_with_fail = f"{config.url}{sep}fail"
-            _http_request(url_with_fail)
+            _http_request(config.fail_url)
             raise e
 
     return t.cast(WrappedFn, healthcheck_wrapper)

tests/test_decorator.py

@@ -8,7 +8,6 @@
 import pytest
 
 from healthchecks_decorator import healthcheck
-from healthchecks_decorator.decorator import _http_request
 from healthchecks_decorator.decorator import HealthcheckConfig
 
 
@@ -26,7 +25,6 @@ def function_to_wrap() -> bool:
         return True
 
     with patch("healthchecks_decorator.decorator.urlopen") as urlopen_mock:
-
         assert function_to_wrap()
         urlopen_mock.assert_called_once_with(url, data=None, timeout=10)
 
@@ -121,10 +119,49 @@ def f(diag: t.Any) -> t.Any:
         urlopen_mock.assert_called_once_with(url, data=None, timeout=10)
 
 
-def test_wrong_url_schema() -> None:
+def test_url_with_query() -> None:
+    """Test urls with queries, as '?create=1'."""
+    slug_url = "https://hc-ping.com/fqOOd6-F4MMNuCEnzTU01w/db-backups?create=1"
+    c = HealthcheckConfig(url=slug_url, send_start=True, send_diagnostics=False)
+    assert c.url == slug_url
+    assert (
+        c.start_url
+        == "https://hc-ping.com/fqOOd6-F4MMNuCEnzTU01w/db-backups/start?create=1"
+    )
+    assert (
+        c.fail_url
+        == "https://hc-ping.com/fqOOd6-F4MMNuCEnzTU01w/db-backups/fail?create=1"
+    )
+
+
+def test_invalid_url() -> None:
     """Test invalid URL schemas."""
-    with pytest.raises(ValueError):
-        _http_request("file:///tmp/localfile.txt")
+    args = dict(send_start=True, send_diagnostics=False)
+
+    # Valid URL
+    assert (
+        bool(HealthcheckConfig(url="https://fake-hc.com/0000-1111-2222-3333", **args))
+        is True
+    )
+
+    # Empty or None URL
+    assert bool(HealthcheckConfig(url="", **args)) is False
+    assert bool(HealthcheckConfig(url=None, **args)) is False
+
+    # Non-HTTP(S) scheme
+    assert (
+        bool(HealthcheckConfig(url="ftp://fake-hc.com/0000-1111-2222-3333", **args))
+        is False
+    )
+
+    # No scheme
+    assert bool(HealthcheckConfig(url="dkakasdkjdjakdjadjfalskdjfalk", **args)) is False
+
+    # No netloc
+    assert bool(HealthcheckConfig(url="https://", **args)) is False
+
+    # Wrong type
+    assert bool(HealthcheckConfig(url=123.23, **args)) is False  # type: ignore
 
 
 def test_envvars(url: str) -> None: