Catch foreign key constraint errors in PiccoloCRUD (#244)

* catch foreign key violation

* add a test

piccolo_api/crud/endpoints.py

@@ -1203,6 +1203,7 @@ async def patch_single(
                 )
 
     @apply_validators
+    @db_exception_handler
     async def delete_single(
         self, request: Request, row_id: PK_TYPES
     ) -> Response:

piccolo_api/crud/exceptions.py

@@ -8,9 +8,16 @@
 try:
     # We can't be sure that asyncpg is installed, hence why it's in a
     # try / except.
-    from asyncpg.exceptions import NotNullViolationError, UniqueViolationError
+    from asyncpg.exceptions import (
+        ForeignKeyViolationError,
+        NotNullViolationError,
+        UniqueViolationError,
+    )
 except ImportError:
 
+    class ForeignKeyViolationError(Exception):  # type: ignore
+        pass
+
     class UniqueViolationError(Exception):  # type: ignore
         pass
 
@@ -75,5 +82,16 @@ async def inner(*args, **kwargs):
                 },
                 status_code=422,
             )
+        except ForeignKeyViolationError as exception:
+            # This is raised when we have an `ON DELETE RESTRICT` constraint
+            # on a foreign key, which prevents us from deleting a row if it's
+            # referenced by a foreign key.
+            logger.exception("Asyncpg foreign key violation")
+            return JSONResponse(
+                {
+                    "db_error": exception.message,
+                },
+                status_code=422,
+            )
 
     return inner

tests/crud/test_crud_endpoints.py

@@ -3,8 +3,9 @@
 
 from piccolo.apps.user.tables import BaseUser
 from piccolo.columns import Email, ForeignKey, Integer, Secret, Text, Varchar
+from piccolo.columns.column_types import OnDelete
 from piccolo.columns.readable import Readable
-from piccolo.table import Table
+from piccolo.table import Table, create_db_tables_sync, drop_db_tables_sync
 from starlette.datastructures import QueryParams
 from starlette.testclient import TestClient
 
@@ -1346,6 +1347,50 @@ def test_put(self):
         self.assertEqual(response.status_code, 204)
 
 
+class TestForeignKeyViolationException(TestCase):
+    """
+    Make sure that if a foreign key violation is raised, we get a useful
+    message back, and not a 500 error. Implemented by the
+    ``@db_exception_handler`` decorator.
+    """
+
+    def setUp(self):
+        class Director(Table):
+            name = Varchar()
+
+        class Movie(Table):
+            name = Varchar()
+            director = ForeignKey(
+                references=Director, on_delete=OnDelete.restrict
+            )
+
+        self.table_classes = (Director, Movie)
+
+        create_db_tables_sync(*self.table_classes, if_not_exists=True)
+
+        self.director = Director({Director.name: "George Lucas"})
+        self.director.save().run_sync()
+
+        Movie(
+            {Movie.director: self.director, Movie.name: "Star Wars"}
+        ).save().run_sync()
+
+    def tearDown(self):
+        drop_db_tables_sync(*self.table_classes)
+
+    def test_delete(self):
+        director = self.director
+
+        client = TestClient(
+            PiccoloCRUD(table=self.director.__class__, read_only=False)
+        )
+
+        # Test error
+        response = client.delete(f"/{director.id}/")
+        self.assertEqual(response.status_code, 422)
+        self.assertIn("db_error", response.json())
+
+
 class TestGet(TestCase):
     def setUp(self):
         for table in (Movie, Role):