diff --git a/airflow/api/client/local_client.py b/airflow/api/client/local_client.py
index 39680e186e86f..c0050672a8e47 100644
--- a/airflow/api/client/local_client.py
+++ b/airflow/api/client/local_client.py
@@ -49,6 +49,9 @@ def get_pools(self):
     def create_pool(self, name, slots, description):
         if not (name and name.strip()):
             raise AirflowBadRequest("Pool name shouldn't be empty")
+        pool_name_length = Pool.pool.property.columns[0].type.length
+        if len(name) > pool_name_length:
+            raise AirflowBadRequest(f"pool name cannot be more than {pool_name_length} characters")
         try:
             slots = int(slots)
         except ValueError:
diff --git a/tests/api/client/test_local_client.py b/tests/api/client/test_local_client.py
index a351539a4117f..9f574e4fc657a 100644
--- a/tests/api/client/test_local_client.py
+++ b/tests/api/client/test_local_client.py
@@ -24,7 +24,6 @@
 
 import pytest
 from freezegun import freeze_time
-from sqlalchemy.exc import DataError
 
 from airflow.api.client.local_client import Client
 from airflow.example_dags import example_bash_operator
@@ -162,8 +161,11 @@ def test_create_pool_bad_slots(self):
 
     def test_create_pool_name_too_long(self):
         long_name = ''.join(random.choices(string.ascii_lowercase, k=300))
-        with pytest.raises(DataError):
-            Pool.create_or_update_pool(
+        pool_name_length = Pool.pool.property.columns[0].type.length
+        with pytest.raises(
+            AirflowBadRequest, match=f"^pool name cannot be more than {pool_name_length} characters"
+        ):
+            self.client.create_pool(
                 name=long_name,
                 slots=5,
                 description='',