Fix 500 error when database access is denied

Instead, a nicer error message is shown.

src/dso_api/dynamic_api/views/api.py

@@ -13,19 +13,23 @@ class (namely the :class:`~dso_api.dynamic_api.views.DynamicApiViewSet` base cla
 
 from __future__ import annotations
 
+import logging
 from functools import cached_property
 
 from django.db import models
+from django.db.utils import ProgrammingError
 from django.utils.translation import gettext as _
 from rest_framework import viewsets
-from rest_framework.exceptions import NotFound
+from rest_framework.exceptions import NotFound, PermissionDenied
 from schematools.contrib.django.models import DynamicModel
 
 from dso_api.dynamic_api import filters, permissions, serializers
 from dso_api.dynamic_api.temporal import TemporalTableQuery
 from dso_api.dynamic_api.utils import limit_queryset_for_scopes
 from rest_framework_dso.views import DSOViewMixin
 
+logger = logging.getLogger(__name__)
+
 
 class DynamicApiViewSet(DSOViewMixin, viewsets.ReadOnlyModelViewSet):
     """Viewset for an API, that is DSO-compatible and dynamically generated.
@@ -59,6 +63,33 @@ class DynamicApiViewSet(DSOViewMixin, viewsets.ReadOnlyModelViewSet):
     # The 'bronhouder' of the associated dataset
     authorization_grantor: str = None
 
+    def list(self, request, *args, **kwargs):
+        try:
+            return super().list(request, *args, **kwargs)
+        except ProgrammingError as e:
+            self._handle_db_error(e)
+            raise
+
+    def retrieve(self, request, *args, **kwargs):
+        try:
+            return super().retrieve(request, *args, **kwargs)
+        except ProgrammingError as e:
+            self._handle_db_error(e)
+            raise
+
+    def _handle_db_error(self, e: ProgrammingError) -> None:
+        """Make sure database permission errors are gratefully handled.
+        This is a common source of 500 error issues,
+        and giving a better response to the user helps.
+        """
+        if str(e).startswith("permission denied for "):
+            logger.exception("Database role has no access (while application allowed): %s", e)
+            raise PermissionDenied(
+                "Database role has no access to the given tables"
+                " (schema permissions were satisfied).",
+                code="db_permission_denied",
+            ) from e
+
     def initial(self, request, *args, **kwargs):
         super().initial(request, *args, **kwargs)
         table_schema = self.model.table_schema()

src/tests/test_dynamic_api/views/test_api_auth.py

@@ -1,4 +1,6 @@
 import pytest
+from django.db import ProgrammingError
+from django.db.backends.utils import CursorWrapper
 from django.urls import reverse
 from schematools.contrib.django import models
 from schematools.types import ProfileSchema
@@ -542,3 +544,56 @@ def test_filter_auth(
         response = api_client.get(url)
         data = read_response_json(response)
         assert "eigenaarNaam" in data["_embedded"]["containers"][0]
+
+    @pytest.mark.parametrize("page", [1, 2])
+    def test_database_denies_access_list(
+        self, api_client, afval_dataset, page, monkeypatch, filled_router
+    ):
+        """Prove that database access issues are gracefully handled."""
+
+        def _mock_execute(*args, **kwargs):
+            raise ProgrammingError("permission denied for TABLE foobar")
+
+        url = reverse("dynamic_api:afvalwegingen-containers-list")
+        qs_args = {"page": page, "_fields": "id,serienummer"}
+        api_client.get(url, data=qs_args)  # warm up AuthMiddleware
+
+        monkeypatch.setattr(CursorWrapper, "execute", _mock_execute)
+        response = api_client.get(url, data=qs_args)
+        data = read_response_json(response)
+        assert response.status_code == 403, data  # permission denied
+        assert data == {
+            "type": "urn:apiexception:db_permission_denied",
+            "title": "You do not have permission to perform this action.",
+            "detail": (
+                "Database role has no access to the given tables"
+                " (schema permissions were satisfied)."
+            ),
+            "status": 403,
+        }
+
+    def test_database_denies_access_detail(
+        self, api_client, afval_dataset, afval_container, monkeypatch, filled_router
+    ):
+        """Prove that database access issues are gracefully handled."""
+
+        def _mock_execute(*args, **kwargs):
+            raise ProgrammingError("permission denied for TABLE foobar")
+
+        url = reverse("dynamic_api:afvalwegingen-containers-detail", args=[afval_container.pk])
+        qs_args = {"_fields": "id,serienummer"}
+        api_client.get(url, data=qs_args)  # warm up AuthMiddleware
+
+        monkeypatch.setattr(CursorWrapper, "execute", _mock_execute)
+        response = api_client.get(url, data=qs_args)
+        data = read_response_json(response)
+        assert response.status_code == 403, data  # permission denied
+        assert data == {
+            "type": "urn:apiexception:db_permission_denied",
+            "title": "You do not have permission to perform this action.",
+            "detail": (
+                "Database role has no access to the given tables"
+                " (schema permissions were satisfied)."
+            ),
+            "status": 403,
+        }