diff options
| author | Dennis Fink | 2026-08-21 10:37:56 +0200 |
|---|---|---|
| committer | Dennis Fink | 2026-08-21 10:37:56 +0200 |
| commit | c5492398f100ccb154fa557d657bae2322f95cf7 (patch) | |
| tree | 21efa67e00c26ab0733a931e6a5810b4c933388b /webmentions_ssg/tasks/extension.py | |
| parent | 5bfcc3c06ebe6d6694221c25f07a4901572f97f5 (diff) | |
| download | webmentions-ssg-c5492398f100ccb154fa557d657bae2322f95cf7.tar.gz webmentions-ssg-c5492398f100ccb154fa557d657bae2322f95cf7.zip | |
test(core): align tests with module ownership
Move direct validator, model, and Huey extension coverage into dedicated
test modules and mock dependencies where behavior belongs to another
module.
Expand Huey extension typing and coverage while removing duplicate
assertions from scanner, sender, form, and view tests.
Diffstat (limited to 'webmentions_ssg/tasks/extension.py')
| -rw-r--r-- | webmentions_ssg/tasks/extension.py | 102 |
1 files changed, 83 insertions, 19 deletions
diff --git a/webmentions_ssg/tasks/extension.py b/webmentions_ssg/tasks/extension.py index 95f190f..25f8e47 100644 --- a/webmentions_ssg/tasks/extension.py +++ b/webmentions_ssg/tasks/extension.py @@ -2,22 +2,50 @@ # # SPDX-License-Identifier: BSD-3-Clause +from collections.abc import Callable from functools import wraps -from typing import Any, Callable +from typing import Any, ParamSpec, TypeVar from urllib.parse import urlsplit, urlunsplit from flask import Flask +from huey import Huey as BaseHuey +from huey.api import TaskWrapper + +P = ParamSpec("P") +R = TypeVar("R") class Huey: - def __init__(self, app: Flask | None = None): + """ + Provide Flask integration for a Huey instance. + + The extension initializes a Huey backend from the Flask configuration and + wraps tasks so that they execute within an application context. + """ + + def __init__(self, app: Flask | None = None) -> None: + """ + Initialize the Huey extension. + + :param app: Flask application to initialize immediately, if provided. + """ self.app: Flask | None = None - self._huey = None + self._huey: BaseHuey | None = None if app is not None: self.init_app(app) - def init_app(self, app: Flask): + def init_app(self, app: Flask) -> None: + """ + Initialize Huey for a Flask application. + + The backend and its storage options are derived from ``HUEY_URL`` and the + resulting extension is registered with the application. + + :param app: Flask application to initialize. + :raises RuntimeError: If the configured Huey backend URL is invalid or uses + an unsupported scheme. + """ config: dict[str, Any] = { "name": app.import_name, "results": True, @@ -36,7 +64,13 @@ class Huey: app.extensions["huey"] = self @property - def huey(self): + def huey(self) -> BaseHuey: + """ + Return the initialized Huey instance. + + :return: Configured Huey backend instance. + :raises RuntimeError: If the extension has not been initialized. + """ if self._huey is None: raise RuntimeError( "Huey has not been initialized. " @@ -44,10 +78,20 @@ class Huey: ) return self._huey - def task(self, *task_args: Any, **task_kwargs: Any): - def decorator(func: Callable): + def task( + self, *task_args: Any, **task_kwargs: Any + ) -> Callable[[Callable[P, object]], TaskWrapper]: + """ + Create a Huey task that runs within the Flask application context. + + :param task_args: Positional arguments forwarded to Huey's task decorator. + :param task_kwargs: Keyword arguments forwarded to Huey's task decorator. + :return: Decorator that registers the wrapped function as a Huey task. + """ + + def decorator(func: Callable[P, R]) -> TaskWrapper: @wraps(func) - def wrapper(*args: Any, **kwargs: Any): + def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: if self.app is None: raise RuntimeError("Flask app is not available.") @@ -58,10 +102,23 @@ class Huey: return decorator - def periodic_task(self, *task_args: Any, **task_kwargs: Any): - def decorator(func: Callable): + def periodic_task( + self, *task_args: Any, **task_kwargs: Any + ) -> Callable[[Callable[P, object]], TaskWrapper]: + """ + Create a periodic Huey task that runs within the Flask application context. + + :param task_args: Positional arguments forwarded to Huey's periodic task + decorator. + :param task_kwargs: Keyword arguments forwarded to Huey's periodic task + decorator. + :return: Decorator that registers the wrapped function as a periodic Huey + task. + """ + + def decorator(func: Callable[P, R]) -> TaskWrapper: @wraps(func) - def wrapper(*args: Any, **kwargs: Any): + def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: if self.app is None: raise RuntimeError("Flask app is not available.") @@ -72,23 +129,30 @@ class Huey: return decorator - def __getattr__(self, name: str): + def __getattr__(self, name: str) -> Any: """ - Forward unknown attributes to the real Huey instance. + Forward an unknown attribute to the underlying Huey instance. - This lets you still use things like: - huey.enqueue(...) - huey.scheduled() - huey.pending() + :param name: Name of the attribute to retrieve. + :return: Attribute from the initialized Huey instance. + :raises RuntimeError: If the extension has not been initialized. """ return getattr(self.huey, name) @staticmethod - def backend_from_url(url: str) -> tuple[Any, dict[str, str]]: + def backend_from_url(url: str) -> tuple[type[BaseHuey], dict[str, Any]]: + """ + Determine the Huey backend and storage options from a URL. + + :param url: Huey backend URL. + :return: Huey backend class and keyword arguments for its storage backend. + :raises RuntimeError: If the URL is malformed for the selected backend or + uses an unsupported scheme. + """ parsed = urlsplit(url) scheme = parsed.scheme.lower() - if scheme.startswith("redis") or scheme.startswith("rediss"): + if scheme.startswith(("redis", "rediss")): fixed_url = urlunsplit(parsed._replace(scheme=scheme.split("+", 1)[0])) if scheme.endswith("priority+expire"): |
