fix(proxy): handle missing prisma in db-less auth

This commit is contained in:
jeevan6996 2026-08-31 19:11:02 +01:00
parent c03a38d501
commit 27e810ea82
2 changed files with 53 additions and 19 deletions

View file

@ -55,13 +55,17 @@ class PrismaDBExceptionHandler:
Reporting decisions want the opposite breadth; use
``is_database_infrastructure_error`` for those.
"""
import prisma.engine.errors
if isinstance(e, DB_CONNECTION_ERROR_TYPES):
return True
if isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection:
return True
try:
import prisma.engine.errors
except ImportError:
return False
if isinstance(e, prisma.engine.errors.EngineConnectionError):
return True
return isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection
return False
@staticmethod
def is_database_infrastructure_error(e: Exception) -> bool:
@ -79,8 +83,14 @@ class PrismaDBExceptionHandler:
``RecordNotFoundError``, etc.) are excluded — the DB IS reachable and
the request itself is what failed.
"""
import prisma
if isinstance(e, DB_CONNECTION_ERROR_TYPES):
return True
if isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection:
return True
try:
import prisma
except ImportError:
return False
data_layer_errors: Final = (
prisma.errors.DataError,
prisma.errors.UniqueViolationError,
@ -92,12 +102,8 @@ class PrismaDBExceptionHandler:
)
if isinstance(e, data_layer_errors):
return False
if isinstance(e, DB_CONNECTION_ERROR_TYPES):
return True
if isinstance(e, prisma.errors.PrismaError):
return True
if isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection:
return True
return False
@staticmethod
@ -119,8 +125,10 @@ class PrismaDBExceptionHandler:
per-row data rejection has to additionally consult
``is_database_service_unavailable_error`` before acting on a True here.
"""
import prisma
try:
import prisma
except ImportError:
return False
return type(e) is prisma.errors.DataError
@staticmethod
@ -132,10 +140,14 @@ class PrismaDBExceptionHandler:
Use this for reconnect logic — data-layer errors like UniqueViolationError
mean the DB IS reachable, so reconnecting would be pointless.
"""
import prisma
if isinstance(e, DB_CONNECTION_ERROR_TYPES):
return True
if isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection:
return True
try:
import prisma
except ImportError:
return False
if isinstance(
e,
(
@ -162,15 +174,15 @@ class PrismaDBExceptionHandler:
)
if any(keyword in error_message for keyword in connection_keywords):
return True
if isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection:
return True
return False
@staticmethod
def is_deadlock_error(e: Exception) -> bool:
"""True iff ``e`` is a Postgres deadlock (P2034 / 40P01) surfaced through prisma."""
import prisma
try:
import prisma
except ImportError:
return False
if not isinstance(e, prisma.errors.PrismaError):
return False
if getattr(e, "code", None) == "P2034":
@ -200,8 +212,10 @@ class PrismaDBExceptionHandler:
are already classified by type/keyword above, and data-layer ones
(the DB IS reachable) must stay 401.
"""
import prisma
try:
import prisma
except ImportError:
return False
if isinstance(e, prisma.errors.PrismaError):
return False
tb = getattr(e, "__traceback__", None)

View file

@ -106,6 +106,26 @@ def test_is_database_connection_generic_errors():
)
def test_db_less_proxy_auth_error_does_not_require_prisma():
"""A missing Prisma dependency must not mask a normal authentication error."""
original_import = __import__
def import_without_prisma(name, globals=None, locals=None, fromlist=(), level=0):
if name == "prisma" or name.startswith("prisma."):
raise ModuleNotFoundError("No module named 'prisma'")
return original_import(name, globals, locals, fromlist, level)
with patch("builtins.__import__", side_effect=import_without_prisma):
error = Exception("No api key passed in.")
assert PrismaDBExceptionHandler.is_database_connection_error(error) is False
assert PrismaDBExceptionHandler.is_database_infrastructure_error(error) is False
assert PrismaDBExceptionHandler.is_prisma_data_error(error) is False
assert PrismaDBExceptionHandler.is_database_transport_error(error) is False
assert PrismaDBExceptionHandler.is_deadlock_error(error) is False
assert PrismaDBExceptionHandler.is_prisma_engine_internal_error(error) is False
assert PrismaDBExceptionHandler.is_database_service_unavailable_error(error) is False
@pytest.mark.parametrize(
"error",
[