Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/exc.py: 70%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# exc.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""Exceptions used with SQLAlchemy.
10The base exception class is :exc:`.SQLAlchemyError`. Exceptions which are
11raised as a result of DBAPI exceptions are all subclasses of
12:exc:`.DBAPIError`.
14"""
15from __future__ import annotations
17import typing
18from typing import Any
19from typing import List
20from typing import Optional
21from typing import overload
22from typing import Tuple
23from typing import Type
24from typing import Union
26from .util import compat
27from .util import preloaded as _preloaded
29if typing.TYPE_CHECKING:
30 from .engine.interfaces import _AnyExecuteParams
31 from .engine.interfaces import Dialect
32 from .sql.compiler import Compiled
33 from .sql.compiler import TypeCompiler
34 from .sql.elements import ClauseElement
36if typing.TYPE_CHECKING:
37 _version_token: str
38else:
39 # set by __init__.py
40 _version_token = None
43class HasDescriptionCode:
44 """helper which adds 'code' as an attribute and '_code_str' as a method"""
46 code: Optional[str] = None
48 def __init__(self, *arg: Any, **kw: Any):
49 code = kw.pop("code", None)
50 if code is not None:
51 self.code = code
52 super().__init__(*arg, **kw)
54 _what_are_we = "error"
56 def _code_str(self) -> str:
57 if not self.code:
58 return ""
59 else:
60 return (
61 f"(Background on this {self._what_are_we} at: "
62 f"https://sqlalche.me/e/{_version_token}/{self.code})"
63 )
65 def __str__(self) -> str:
66 message = super().__str__()
67 if self.code:
68 message = "%s %s" % (message, self._code_str())
69 return message
72class SQLAlchemyError(HasDescriptionCode, Exception):
73 """Generic error class."""
75 def _message(self) -> str:
76 # rules:
77 #
78 # 1. single arg string will usually be a unicode
79 # object, but since __str__() must return unicode, check for
80 # bytestring just in case
81 #
82 # 2. for multiple self.args, this is not a case in current
83 # SQLAlchemy though this is happening in at least one known external
84 # library, call str() which does a repr().
85 #
86 text: str
88 if len(self.args) == 1:
89 arg_text = self.args[0]
91 if isinstance(arg_text, bytes):
92 text = compat.decode_backslashreplace(arg_text, "utf-8")
93 # This is for when the argument is not a string of any sort.
94 # Otherwise, converting this exception to string would fail for
95 # non-string arguments.
96 else:
97 text = str(arg_text)
99 return text
100 else:
101 # this is not a normal case within SQLAlchemy but is here for
102 # compatibility with Exception.args - the str() comes out as
103 # a repr() of the tuple
104 return str(self.args)
106 def _sql_message(self) -> str:
107 message = self._message()
109 if self.code:
110 message = "%s %s" % (message, self._code_str())
112 return message
114 def __str__(self) -> str:
115 return self._sql_message()
118class ArgumentError(SQLAlchemyError):
119 """Raised when an invalid or conflicting function argument is supplied.
121 This error generally corresponds to construction time state errors.
123 """
126class DuplicateColumnError(ArgumentError):
127 """a Column is being added to a Table that would replace another
128 Column, without appropriate parameters to allow this in place.
130 .. versionadded:: 2.0.0b4
132 """
135class ObjectNotExecutableError(ArgumentError):
136 """Raised when an object is passed to .execute() that can't be
137 executed as SQL.
139 """
141 def __init__(self, target: Any):
142 super().__init__(f"Not an executable object: {target!r}")
143 self.target = target
145 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
146 return self.__class__, (self.target,)
149class NoSuchModuleError(ArgumentError):
150 """Raised when a dynamically-loaded module (usually a database dialect)
151 of a particular name cannot be located."""
154class NoForeignKeysError(ArgumentError):
155 """Raised when no foreign keys can be located between two selectables
156 during a join."""
159class AmbiguousForeignKeysError(ArgumentError):
160 """Raised when more than one foreign key matching can be located
161 between two selectables during a join."""
164class ConstraintColumnNotFoundError(ArgumentError):
165 """raised when a constraint refers to a string column name that
166 is not present in the table being constrained.
168 .. versionadded:: 2.0
170 """
173class CircularDependencyError(SQLAlchemyError):
174 """Raised by topological sorts when a circular dependency is detected.
176 There are two scenarios where this error occurs:
178 * In a Session flush operation, if two objects are mutually dependent
179 on each other, they can not be inserted or deleted via INSERT or
180 DELETE statements alone; an UPDATE will be needed to post-associate
181 or pre-deassociate one of the foreign key constrained values.
182 The ``post_update`` flag described at :ref:`post_update` can resolve
183 this cycle.
184 * In a :attr:`_schema.MetaData.sorted_tables` operation, two
185 :class:`_schema.ForeignKey`
186 or :class:`_schema.ForeignKeyConstraint` objects mutually refer to each
187 other. Apply the ``use_alter=True`` flag to one or both,
188 see :ref:`use_alter`.
190 """
192 def __init__(
193 self,
194 message: str,
195 cycles: Any,
196 edges: Any,
197 msg: Optional[str] = None,
198 code: Optional[str] = None,
199 ):
200 if msg is None:
201 message += " (%s)" % ", ".join(repr(s) for s in cycles)
202 else:
203 message = msg
204 SQLAlchemyError.__init__(self, message, code=code)
205 self.cycles = cycles
206 self.edges = edges
208 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
209 return (
210 self.__class__,
211 (None, self.cycles, self.edges, self.args[0]),
212 {"code": self.code} if self.code is not None else {},
213 )
216class CompileError(SQLAlchemyError):
217 """Raised when an error occurs during SQL compilation"""
220class UnsupportedCompilationError(CompileError):
221 """Raised when an operation is not supported by the given compiler.
223 .. seealso::
225 :ref:`faq_sql_expression_string`
227 :ref:`error_l7de`
228 """
230 code = "l7de"
232 def __init__(
233 self,
234 compiler: Union[Compiled, TypeCompiler],
235 element_type: Type[ClauseElement],
236 message: Optional[str] = None,
237 ):
238 super().__init__(
239 "Compiler %r can't render element of type %s%s"
240 % (compiler, element_type, ": %s" % message if message else "")
241 )
242 self.compiler = compiler
243 self.element_type = element_type
244 self.message = message
246 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
247 return self.__class__, (self.compiler, self.element_type, self.message)
250class IdentifierError(SQLAlchemyError):
251 """Raised when a schema name is beyond the max character limit"""
254class DisconnectionError(SQLAlchemyError):
255 """A disconnect is detected on a raw DB-API connection.
257 This error is raised and consumed internally by a connection pool. It can
258 be raised by the :meth:`_events.PoolEvents.checkout`
259 event so that the host pool
260 forces a retry; the exception will be caught three times in a row before
261 the pool gives up and raises :class:`~sqlalchemy.exc.InvalidRequestError`
262 regarding the connection attempt.
264 """
266 invalidate_pool: bool = False
269class InvalidatePoolError(DisconnectionError):
270 """Raised when the connection pool should invalidate all stale connections.
272 A subclass of :class:`_exc.DisconnectionError` that indicates that the
273 disconnect situation encountered on the connection probably means the
274 entire pool should be invalidated, as the database has been restarted.
276 This exception will be handled otherwise the same way as
277 :class:`_exc.DisconnectionError`, allowing three attempts to reconnect
278 before giving up.
280 """
282 invalidate_pool: bool = True
285class TimeoutError(SQLAlchemyError): # noqa
286 """Raised when a connection pool times out on getting a connection."""
289class InvalidRequestError(SQLAlchemyError):
290 """SQLAlchemy was asked to do something it can't do.
292 This error generally corresponds to runtime state errors.
294 """
297class IllegalStateChangeError(InvalidRequestError):
298 """An object that tracks state encountered an illegal state change
299 of some kind.
301 .. versionadded:: 2.0
303 """
306class NoInspectionAvailable(InvalidRequestError):
307 """A subject passed to :func:`sqlalchemy.inspection.inspect` produced
308 no context for inspection."""
311class PendingRollbackError(InvalidRequestError):
312 """A transaction has failed and needs to be rolled back before
313 continuing.
315 .. versionadded:: 1.4
317 """
320class ResourceClosedError(InvalidRequestError):
321 """An operation was requested from a connection, cursor, or other
322 object that's in a closed state."""
325class NoSuchColumnError(InvalidRequestError, KeyError):
326 """A nonexistent column is requested from a ``Row``."""
329class NoResultFound(InvalidRequestError):
330 """A database result was required but none was found.
333 .. versionchanged:: 1.4 This exception is now part of the
334 ``sqlalchemy.exc`` module in Core, moved from the ORM. The symbol
335 remains importable from ``sqlalchemy.orm.exc``.
338 """
341class MultipleResultsFound(InvalidRequestError):
342 """A single database result was required but more than one were found.
344 .. versionchanged:: 1.4 This exception is now part of the
345 ``sqlalchemy.exc`` module in Core, moved from the ORM. The symbol
346 remains importable from ``sqlalchemy.orm.exc``.
349 """
352class NoReferenceError(InvalidRequestError):
353 """Raised by ``ForeignKey`` to indicate a reference cannot be resolved."""
355 table_name: str
358class AwaitRequired(InvalidRequestError):
359 """Error raised by the async greenlet spawn if no async operation
360 was awaited when it required one.
362 """
364 code = "xd1r"
367class MissingGreenlet(InvalidRequestError):
368 r"""Error raised by the async greenlet await\_ if called while not inside
369 the greenlet spawn context.
371 """
373 code = "xd2s"
376class NoReferencedTableError(NoReferenceError):
377 """Raised by ``ForeignKey`` when the referred ``Table`` cannot be
378 located.
380 """
382 def __init__(self, message: str, tname: str):
383 NoReferenceError.__init__(self, message)
384 self.table_name = tname
386 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
387 return self.__class__, (self.args[0], self.table_name)
390class NoReferencedColumnError(NoReferenceError):
391 """Raised by ``ForeignKey`` when the referred ``Column`` cannot be
392 located.
394 """
396 def __init__(self, message: str, tname: str, cname: str):
397 NoReferenceError.__init__(self, message)
398 self.table_name = tname
399 self.column_name = cname
401 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
402 return (
403 self.__class__,
404 (self.args[0], self.table_name, self.column_name),
405 )
408class NoSuchTableError(InvalidRequestError):
409 """Table does not exist or is not visible to a connection."""
412class UnreflectableTableError(InvalidRequestError):
413 """Table exists but can't be reflected for some reason."""
416class UnboundExecutionError(InvalidRequestError):
417 """SQL was attempted without a database connection to execute it on."""
420class DontWrapMixin:
421 """A mixin class which, when applied to a user-defined Exception class,
422 will not be wrapped inside of :exc:`.StatementError` if the error is
423 emitted within the process of executing a statement.
425 E.g.::
427 from sqlalchemy.exc import DontWrapMixin
430 class MyCustomException(Exception, DontWrapMixin):
431 pass
434 class MySpecialType(TypeDecorator):
435 impl = String
437 def process_bind_param(self, value, dialect):
438 if value == "invalid":
439 raise MyCustomException("invalid!")
441 """
444class StatementError(SQLAlchemyError):
445 """An error occurred during execution of a SQL statement.
447 :class:`StatementError` wraps the exception raised
448 during execution, and features :attr:`.statement`
449 and :attr:`.params` attributes which supply context regarding
450 the specifics of the statement which had an issue.
452 The wrapped exception object is available in
453 the :attr:`.orig` attribute.
455 """
457 statement: Optional[str] = None
458 """The string SQL statement being invoked when this exception occurred."""
460 params: Optional[_AnyExecuteParams] = None
461 """The parameter list being used when this exception occurred."""
463 orig: Optional[BaseException] = None
464 """The original exception that was thrown.
466 """
468 ismulti: Optional[bool] = None
469 """multi parameter passed to repr_params(). None is meaningful."""
471 connection_invalidated: bool = False
473 def __init__(
474 self,
475 message: str,
476 statement: Optional[str],
477 params: Optional[_AnyExecuteParams],
478 orig: Optional[BaseException],
479 hide_parameters: bool = False,
480 code: Optional[str] = None,
481 ismulti: Optional[bool] = None,
482 ):
483 SQLAlchemyError.__init__(self, message, code=code)
484 self.statement = statement
485 self.params = params
486 self.orig = orig
487 self.ismulti = ismulti
488 self.hide_parameters = hide_parameters
489 self.detail: List[str] = []
491 def add_detail(self, msg: str) -> None:
492 self.detail.append(msg)
494 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
495 return (
496 self.__class__,
497 (
498 self.args[0],
499 self.statement,
500 self.params,
501 self.orig,
502 self.hide_parameters,
503 self.__dict__.get("code"),
504 self.ismulti,
505 ),
506 {"detail": self.detail},
507 )
509 @_preloaded.preload_module("sqlalchemy.sql.util")
510 def _sql_message(self) -> str:
511 util = _preloaded.sql_util
513 details = [self._message()]
514 if self.statement:
515 stmt_detail = "[SQL: %s]" % self.statement
516 details.append(stmt_detail)
517 if self.params:
518 if self.hide_parameters:
519 details.append(
520 "[SQL parameters hidden due to hide_parameters=True]"
521 )
522 else:
523 params_repr = util._repr_params(
524 self.params, 10, ismulti=self.ismulti
525 )
526 details.append("[parameters: %r]" % params_repr)
527 code_str = self._code_str()
528 if code_str:
529 details.append(code_str)
530 return "\n".join(["(%s)" % det for det in self.detail] + details)
533class DBAPIError(StatementError):
534 """Raised when the execution of a database operation fails.
536 Wraps exceptions raised by the DB-API underlying the
537 database operation. Driver-specific implementations of the standard
538 DB-API exception types are wrapped by matching sub-types of SQLAlchemy's
539 :class:`DBAPIError` when possible. DB-API's ``Error`` type maps to
540 :class:`DBAPIError` in SQLAlchemy, otherwise the names are identical. Note
541 that there is no guarantee that different DB-API implementations will
542 raise the same exception type for any given error condition.
544 :class:`DBAPIError` features :attr:`~.StatementError.statement`
545 and :attr:`~.StatementError.params` attributes which supply context
546 regarding the specifics of the statement which had an issue, for the
547 typical case when the error was raised within the context of
548 emitting a SQL statement.
550 The wrapped exception object is available in the
551 :attr:`~.StatementError.orig` attribute. Its type and properties are
552 DB-API implementation specific.
554 """
556 code = "dbapi"
558 @overload
559 @classmethod
560 def instance(
561 cls,
562 statement: Optional[str],
563 params: Optional[_AnyExecuteParams],
564 orig: Exception,
565 dbapi_base_err: Type[Exception],
566 hide_parameters: bool = False,
567 connection_invalidated: bool = False,
568 dialect: Optional[Dialect] = None,
569 ismulti: Optional[bool] = None,
570 ) -> StatementError: ...
572 @overload
573 @classmethod
574 def instance(
575 cls,
576 statement: Optional[str],
577 params: Optional[_AnyExecuteParams],
578 orig: DontWrapMixin,
579 dbapi_base_err: Type[Exception],
580 hide_parameters: bool = False,
581 connection_invalidated: bool = False,
582 dialect: Optional[Dialect] = None,
583 ismulti: Optional[bool] = None,
584 ) -> DontWrapMixin: ...
586 @overload
587 @classmethod
588 def instance(
589 cls,
590 statement: Optional[str],
591 params: Optional[_AnyExecuteParams],
592 orig: BaseException,
593 dbapi_base_err: Type[Exception],
594 hide_parameters: bool = False,
595 connection_invalidated: bool = False,
596 dialect: Optional[Dialect] = None,
597 ismulti: Optional[bool] = None,
598 ) -> BaseException: ...
600 @classmethod
601 def instance(
602 cls,
603 statement: Optional[str],
604 params: Optional[_AnyExecuteParams],
605 orig: Union[BaseException, DontWrapMixin],
606 dbapi_base_err: Type[Exception],
607 hide_parameters: bool = False,
608 connection_invalidated: bool = False,
609 dialect: Optional[Dialect] = None,
610 ismulti: Optional[bool] = None,
611 ) -> Union[BaseException, DontWrapMixin]:
612 # Don't ever wrap these, just return them directly as if
613 # DBAPIError didn't exist.
614 if (
615 isinstance(orig, BaseException) and not isinstance(orig, Exception)
616 ) or isinstance(orig, DontWrapMixin):
617 return orig
619 if orig is not None:
620 # not a DBAPI error, statement is present.
621 # raise a StatementError
622 if isinstance(orig, SQLAlchemyError) and statement:
623 return StatementError(
624 "(%s.%s) %s"
625 % (
626 orig.__class__.__module__,
627 orig.__class__.__name__,
628 orig.args[0],
629 ),
630 statement,
631 params,
632 orig,
633 hide_parameters=hide_parameters,
634 code=orig.code,
635 ismulti=ismulti,
636 )
637 elif not isinstance(orig, dbapi_base_err) and statement:
638 return StatementError(
639 "(%s.%s) %s"
640 % (
641 orig.__class__.__module__,
642 orig.__class__.__name__,
643 orig,
644 ),
645 statement,
646 params,
647 orig,
648 hide_parameters=hide_parameters,
649 ismulti=ismulti,
650 )
652 glob = globals()
653 for super_ in orig.__class__.__mro__:
654 name = super_.__name__
655 if dialect:
656 name = dialect.dbapi_exception_translation_map.get(
657 name, name
658 )
659 if name in glob and issubclass(glob[name], DBAPIError):
660 cls = glob[name]
661 break
663 return cls(
664 statement,
665 params,
666 orig,
667 connection_invalidated=connection_invalidated,
668 hide_parameters=hide_parameters,
669 code=cls.code,
670 ismulti=ismulti,
671 )
673 def __reduce__(self) -> Union[str, Tuple[Any, ...]]:
674 return (
675 self.__class__,
676 (
677 self.statement,
678 self.params,
679 self.orig,
680 self.hide_parameters,
681 self.connection_invalidated,
682 self.__dict__.get("code"),
683 self.ismulti,
684 ),
685 {"detail": self.detail},
686 )
688 def __init__(
689 self,
690 statement: Optional[str],
691 params: Optional[_AnyExecuteParams],
692 orig: BaseException,
693 hide_parameters: bool = False,
694 connection_invalidated: bool = False,
695 code: Optional[str] = None,
696 ismulti: Optional[bool] = None,
697 ):
698 try:
699 text = str(orig)
700 except Exception as e:
701 text = "Error in str() of DB-API-generated exception: " + str(e)
702 StatementError.__init__(
703 self,
704 "(%s.%s) %s"
705 % (orig.__class__.__module__, orig.__class__.__name__, text),
706 statement,
707 params,
708 orig,
709 hide_parameters,
710 code=code,
711 ismulti=ismulti,
712 )
713 self.connection_invalidated = connection_invalidated
716class InterfaceError(DBAPIError):
717 """Wraps a DB-API InterfaceError."""
719 code = "rvf5"
722class DatabaseError(DBAPIError):
723 """Wraps a DB-API DatabaseError."""
725 code = "4xp6"
728class DataError(DatabaseError):
729 """Wraps a DB-API DataError."""
731 code = "9h9h"
734class OperationalError(DatabaseError):
735 """Wraps a DB-API OperationalError."""
737 code = "e3q8"
740class IntegrityError(DatabaseError):
741 """Wraps a DB-API IntegrityError."""
743 code = "gkpj"
746class InternalError(DatabaseError):
747 """Wraps a DB-API InternalError."""
749 code = "2j85"
752class ProgrammingError(DatabaseError):
753 """Wraps a DB-API ProgrammingError."""
755 code = "f405"
758class NotSupportedError(DatabaseError):
759 """Wraps a DB-API NotSupportedError."""
761 code = "tw8g"
764# Warnings
767class SATestSuiteWarning(Warning):
768 """warning for a condition detected during tests that is non-fatal
770 Currently outside of SAWarning so that we can work around tools like
771 Alembic doing the wrong thing with warnings.
773 """
776class SADeprecationWarning(HasDescriptionCode, DeprecationWarning):
777 """Issued for usage of deprecated APIs."""
779 deprecated_since: Optional[str] = None
780 "Indicates the version that started raising this deprecation warning"
783class Base20DeprecationWarning(SADeprecationWarning):
784 """Issued for usage of APIs specifically deprecated or legacy in
785 SQLAlchemy 2.0.
787 .. seealso::
789 :ref:`error_b8d9`.
791 :ref:`deprecation_20_mode`
793 """
795 deprecated_since: Optional[str] = "1.4"
796 "Indicates the version that started raising this deprecation warning"
798 def __str__(self) -> str:
799 return (
800 super().__str__()
801 + " (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)"
802 )
805class LegacyAPIWarning(Base20DeprecationWarning):
806 """indicates an API that is in 'legacy' status, a long term deprecation."""
809class MovedIn20Warning(Base20DeprecationWarning):
810 """Subtype of Base20DeprecationWarning to indicate an API that moved
811 only.
812 """
815class SAPendingDeprecationWarning(PendingDeprecationWarning):
816 """A similar warning as :class:`_exc.SADeprecationWarning`, this warning
817 is not used in modern versions of SQLAlchemy.
819 """
821 deprecated_since: Optional[str] = None
822 "Indicates the version that started raising this deprecation warning"
825class SAWarning(HasDescriptionCode, RuntimeWarning):
826 """Issued at runtime."""
828 _what_are_we = "warning"