integreat_cms/api/decorators.py
"""
This module includes functions that are used as decorators in the API endpoints.
"""
from __future__ import annotations
import json
import logging
import random
import re
import threading
from functools import wraps
from typing import TYPE_CHECKING
from urllib import error, parse, request
from django.conf import settings
from django.http import Http404, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from ..cms.constants import feedback_ratings
from ..cms.models import Language, Region
if TYPE_CHECKING:
from collections.abc import Callable
from typing import Any
from django.http import HttpRequest, HttpResponseRedirect
logger = logging.getLogger(__name__)
def feedback_handler(func: Callable) -> Callable:
"""
Decorator definition for feedback API functions and methods
:param func: decorated function
:return: The decorated feedback view function
"""
@csrf_exempt
@wraps(func)
def handle_feedback(
request: HttpRequest, region_slug: str, language_slug: str
) -> JsonResponse:
"""
Parse feedback API request parameters
:param request: Django request
:param region_slug: slug of a region
:param language_slug: slug of a language
:return: The decorated feedback view function
"""
if request.method != "POST":
return JsonResponse({"error": "Invalid request."}, status=405)
try:
region = Region.objects.get(slug=region_slug)
language = Language.objects.get(slug=language_slug)
except Region.DoesNotExist:
return JsonResponse(
{"error": f'No region found with slug "{region_slug}"'}, status=404
)
except Language.DoesNotExist:
return JsonResponse(
{"error": f'No language found with slug "{language_slug}"'}, status=404
)
data = (
json.loads(request.body.decode())
if request.content_type == "application/json"
else request.POST.dict()
)
comment = data.pop("comment", "")
rating = data.pop("rating", None)
category = data.pop("category", None)
if rating not in [None, "up", "down"]:
return JsonResponse({"error": "Invalid rating."}, status=400)
if not comment and not rating:
return JsonResponse(
{"error": "Either comment or rating is required."}, status=400
)
rating_normalized: bool | None = feedback_ratings.NOT_STATED
if rating == "up":
rating_normalized = feedback_ratings.POSITIVE
elif rating == "down":
rating_normalized = feedback_ratings.NEGATIVE
is_technical = category == "Technisches Feedback"
return func(data, region, language, comment, rating_normalized, is_technical)
return handle_feedback
def json_response(function: Callable) -> Callable:
"""
This decorator can be used to catch :class:`~django.http.Http404` exceptions and convert them to a :class:`~django.http.JsonResponse`.
Without this decorator, the exceptions would be converted to :class:`~django.http.HttpResponse`.
:param function: The view function which should always return JSON
:return: The decorated function
"""
@wraps(function)
def wrap(
request: dict[str, str] | HttpRequest, *args: Any, **kwargs: Any
) -> HttpResponseRedirect | JsonResponse:
r"""
The inner function for this decorator.
It tries to execute the decorated view function and returns the unaltered result with the exception of a
:class:`~django.http.Http404` error, which is converted into JSON format.
:param request: Django request
:param \*args: The supplied arguments
:param \**kwargs: The supplied kwargs
:return: The response of the given function or an 404 :class:`~django.http.JsonResponse`
"""
try:
return function(request, *args, **kwargs)
except Http404 as e:
return JsonResponse({"error": str(e) or "Not found."}, status=404)
return wrap
def matomo_tracking(func: Callable) -> Callable:
"""
This decorator is supposed to be applied to API content endpoints. It will track
the request in Matomo. The request to the Matomo API is executed asynchronously in its
own thread to not block the Integreat CMS API request.
Only the URL and the User Agent will be sent to Matomo.
:param func: decorated function
:return: The decorated feedback view function
"""
def matomo_request(host: str, data: dict) -> None:
"""
Wrap HTTP request to Matomo in threaded function.
:param host: Hostname + protocol of Matomo installation
:param data: Data to send to Matomo API
"""
data_str = parse.urlencode(data)
url = f"{host}/matomo.php?{data_str}"
req = request.Request(url)
try:
with request.urlopen(req):
pass
except error.HTTPError as e:
logger.error(
"Matomo Tracking API request to %r failed with: %s",
# Mask auth token in log
re.sub(r"&token_auth=[^&]+", "&token_auth=********", url),
e,
)
@wraps(func)
def wrap(request: HttpRequest, *args: Any, **kwargs: Any) -> JsonResponse:
r"""
The inner function for this decorator.
:param request: Django request
:param \*args: The supplied arguments
:param \**kwargs: The supplied kwargs
:return: The response of the given function or an 404 :class:`~django.http.JsonResponse`
"""
if (
not settings.MATOMO_TRACKING
or not request.region.matomo_id
or not request.region.matomo_token
):
return func(request, *args, **kwargs)
data = {
"idsite": request.region.matomo_id,
"token_auth": request.region.matomo_token,
"rec": 1,
"url": request.build_absolute_uri(),
"urlref": settings.BASE_URL,
"ua": request.META.get("HTTP_USER_AGENT", "unknown user agent"),
"cip": f"{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}",
}
t = threading.Thread(target=matomo_request, args=(settings.MATOMO_URL, data))
t.daemon = True
t.start()
return func(request, *args, **kwargs)
return wrap