diff --git a/superset/views/base.py b/superset/views/base.py
index 863ca2f84ab6..22e4c5f8d163 100644
--- a/superset/views/base.py
+++ b/superset/views/base.py
@@ -45,7 +45,7 @@
 from flask_wtf.csrf import CSRFError
 from flask_wtf.form import FlaskForm
 from pkg_resources import resource_filename
-from sqlalchemy import or_
+from sqlalchemy import exc, or_
 from sqlalchemy.orm import Query
 from werkzeug.exceptions import HTTPException
 from wtforms import Form
@@ -231,6 +231,9 @@ def wraps(self: "BaseSupersetView", *args: Any, **kwargs: Any) -> FlaskResponse:
             return json_error_response(
                 utils.error_msg_from_exception(ex), status=cast(int, ex.code)
             )
+        except (exc.IntegrityError, exc.DatabaseError, exc.DataError) as ex:
+            logger.exception(ex)
+            return json_error_response(utils.error_msg_from_exception(ex), status=422)
         except Exception as ex:  # pylint: disable=broad-except
             logger.exception(ex)
             return json_error_response(utils.error_msg_from_exception(ex))
diff --git a/superset/views/base_api.py b/superset/views/base_api.py
index 260e5731788b..01b462bb321f 100644
--- a/superset/views/base_api.py
+++ b/superset/views/base_api.py
@@ -39,6 +39,7 @@
 from superset.stats_logger import BaseStatsLogger
 from superset.superset_typing import FlaskResponse
 from superset.utils.core import time_function
+from superset.views.base import handle_api_exception
 
 logger = logging.getLogger(__name__)
 get_related_schema = {
@@ -386,6 +387,7 @@ def send_stats_metrics(
         object_ref=False,
         log_to_statsd=False,
     )
+    @handle_api_exception
     def info_headless(self, **kwargs: Any) -> Response:
         """
         Add statsd metrics to builtin FAB _info endpoint
@@ -399,6 +401,7 @@ def info_headless(self, **kwargs: Any) -> Response:
         object_ref=False,
         log_to_statsd=False,
     )
+    @handle_api_exception
     def get_headless(self, pk: int, **kwargs: Any) -> Response:
         """
         Add statsd metrics to builtin FAB GET endpoint
@@ -412,6 +415,7 @@ def get_headless(self, pk: int, **kwargs: Any) -> Response:
         object_ref=False,
         log_to_statsd=False,
     )
+    @handle_api_exception
     def get_list_headless(self, **kwargs: Any) -> Response:
         """
         Add statsd metrics to builtin FAB GET list endpoint
@@ -425,6 +429,7 @@ def get_list_headless(self, **kwargs: Any) -> Response:
         object_ref=False,
         log_to_statsd=False,
     )
+    @handle_api_exception
     def post_headless(self) -> Response:
         """
         Add statsd metrics to builtin FAB POST endpoint
@@ -438,6 +443,7 @@ def post_headless(self) -> Response:
         object_ref=False,
         log_to_statsd=False,
     )
+    @handle_api_exception
     def put_headless(self, pk: int) -> Response:
         """
         Add statsd metrics to builtin FAB PUT endpoint
@@ -451,6 +457,7 @@ def put_headless(self, pk: int) -> Response:
         object_ref=False,
         log_to_statsd=False,
     )
+    @handle_api_exception
     def delete_headless(self, pk: int) -> Response:
         """
         Add statsd metrics to builtin FAB DELETE endpoint
@@ -464,6 +471,7 @@ def delete_headless(self, pk: int) -> Response:
     @safe
     @statsd_metrics
     @rison(get_related_schema)
+    @handle_api_exception
     def related(self, column_name: str, **kwargs: Any) -> FlaskResponse:
         """Get related fields data
         ---
@@ -542,6 +550,7 @@ def related(self, column_name: str, **kwargs: Any) -> FlaskResponse:
     @safe
     @statsd_metrics
     @rison(get_related_schema)
+    @handle_api_exception
     def distinct(self, column_name: str, **kwargs: Any) -> FlaskResponse:
         """Get distinct values from field data
         ---