From 7ed976410f63e785f0d9482d5e4b90bc65c3efc5 Mon Sep 17 00:00:00 2001
From: Daniel Townsend <dan@dantownsend.co.uk>
Date: Fri, 7 Aug 2026 23:39:55 +0100
Subject: [PATCH 1/2] handle check constraint errors

---
 piccolo_api/crud/exceptions.py | 12 ++++++++++++
 requirements/requirements.txt  |  2 +-
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/piccolo_api/crud/exceptions.py b/piccolo_api/crud/exceptions.py
index b66caf00..e07d08c1 100644
--- a/piccolo_api/crud/exceptions.py
+++ b/piccolo_api/crud/exceptions.py
@@ -9,6 +9,7 @@
     # We can't be sure that asyncpg is installed, hence why it's in a
     # try / except.
     from asyncpg.exceptions import (
+        CheckViolationError,
         ForeignKeyViolationError,
         NotNullViolationError,
         RestrictViolationError,
@@ -16,6 +17,9 @@
     )
 except ImportError:
 
+    class CheckViolationError(Exception):  # type: ignore
+        pass
+
     class RestrictViolationError(Exception):  # type: ignore
         pass
 
@@ -70,6 +74,14 @@ async def inner(*args, **kwargs):
                 {"db_error": exception.__str__()},
                 status_code=422,
             )
+        except CheckViolationError as exception:
+            logger.exception("Asyncpg check violation")
+            return JSONResponse(
+                {
+                    "db_error": exception.message,
+                },
+                status_code=422,
+            )
         except UniqueViolationError as exception:
             logger.exception("Asyncpg unique violation")
             return JSONResponse(
diff --git a/requirements/requirements.txt b/requirements/requirements.txt
index af69a186..130b7d6d 100644
--- a/requirements/requirements.txt
+++ b/requirements/requirements.txt
@@ -1,5 +1,5 @@
 Jinja2>=2.11.0
-piccolo[postgres]>=1.16.0
+piccolo[postgres]>=1.36.0
 pydantic[email]>=2.0
 python-multipart>=0.0.5
 fastapi>=0.100.0

From 263e5cb5fa4f2548fe2e9b6bfd215c00c7fd7634 Mon Sep 17 00:00:00 2001
From: Daniel Townsend <dan@dantownsend.co.uk>
Date: Fri, 7 Aug 2026 23:54:38 +0100
Subject: [PATCH 2/2] Update test_crud_endpoints.py

---
 tests/crud/test_crud_endpoints.py | 98 ++++++++++++++++++++++++++++++-
 1 file changed, 97 insertions(+), 1 deletion(-)

diff --git a/tests/crud/test_crud_endpoints.py b/tests/crud/test_crud_endpoints.py
index e008fd5b..669e5af4 100644
--- a/tests/crud/test_crud_endpoints.py
+++ b/tests/crud/test_crud_endpoints.py
@@ -1,5 +1,5 @@
 from enum import Enum
-from unittest import TestCase
+from unittest import TestCase, skipIf
 
 from piccolo.apps.user.tables import BaseUser
 from piccolo.columns import (
@@ -13,6 +13,7 @@
 )
 from piccolo.columns.column_types import OnDelete
 from piccolo.columns.readable import Readable
+from piccolo.constraints import Check
 from piccolo.table import Table, create_db_tables_sync, drop_db_tables_sync
 from starlette.datastructures import QueryParams
 from starlette.testclient import TestClient
@@ -64,6 +65,13 @@ class Ticket(Table):
     code = Varchar(null=False)
 
 
+class Discount(Table):
+    name = Varchar()
+    percentage = Integer()
+
+    percentage_check = Check((percentage >= 0) & (percentage <= 100))
+
+
 class TestGetVisibleFieldsOptions(TestCase):
     def test_without_joins(self):
         response = get_visible_fields_options(table=Role, max_joins=0)
@@ -1565,6 +1573,94 @@ def test_put(self):
         self.assertEqual(response.status_code, 204)
 
 
+@skipIf(
+    Discount._meta.db.engine_type != "postgres",
+    "Piccolo adds check constraints using `ALTER TABLE`, which SQLite "
+    "doesn't support.",
+)
+class TestCheckException(TestCase):
+    """
+    Make sure that if a check constraint fails, we get a useful message
+    back, and not a 500 error. Implemented by the ``@db_exception_handler``
+    decorator.
+    """
+
+    def setUp(self):
+        Discount.create_table(if_not_exists=True).run_sync()
+
+        self.discount = (
+            Discount.objects()
+            .create(
+                name="Student",
+                percentage=20,
+            )
+            .run_sync()
+        )
+
+    def tearDown(self):
+        Discount.alter().drop_table().run_sync()
+
+    def test_post(self):
+        client = TestClient(PiccoloCRUD(table=Discount, read_only=False))
+
+        # Test error
+        response = client.post(
+            "/",
+            json={"name": "Pensioner", "percentage": 200},
+        )
+        self.assertEqual(response.status_code, 422)
+        self.assertIn("db_error", response.json())
+
+        # Test success
+        response = client.post(
+            "/",
+            json={"name": "Pensioner", "percentage": 50},
+        )
+        self.assertEqual(response.status_code, 201)
+
+    def test_patch(self):
+        client = TestClient(PiccoloCRUD(table=Discount, read_only=False))
+
+        # Test error
+        response = client.patch(
+            f"/{self.discount.id}/",
+            json={"percentage": 200},
+        )
+        self.assertEqual(response.status_code, 422)
+        self.assertIn("db_error", response.json())
+
+        # Test success
+        response = client.patch(
+            f"/{self.discount.id}/",
+            json={"percentage": 50},
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_put(self):
+        client = TestClient(PiccoloCRUD(table=Discount, read_only=False))
+
+        # Test error
+        response = client.put(
+            f"/{self.discount.id}/",
+            json={
+                "name": self.discount.name,
+                "percentage": 200,
+            },
+        )
+        self.assertEqual(response.status_code, 422)
+        self.assertIn("db_error", response.json())
+
+        # Test success
+        response = client.put(
+            f"/{self.discount.id}/",
+            json={
+                "name": self.discount.name,
+                "percentage": 50,
+            },
+        )
+        self.assertEqual(response.status_code, 204)
+
+
 class TestForeignKeyViolationException(TestCase):
     """
     Make sure that if a foreign key violation is raised, we get a useful