diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/exc.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/sqlalchemy/exc.py | 830 |
1 files changed, 0 insertions, 830 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/exc.py b/venv/lib/python3.11/site-packages/sqlalchemy/exc.py deleted file mode 100644 index 7d7eff3..0000000 --- a/venv/lib/python3.11/site-packages/sqlalchemy/exc.py +++ /dev/null @@ -1,830 +0,0 @@ -# exc.py -# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors -# <see AUTHORS file> -# -# This module is part of SQLAlchemy and is released under -# the MIT License: https://www.opensource.org/licenses/mit-license.php - -"""Exceptions used with SQLAlchemy. - -The base exception class is :exc:`.SQLAlchemyError`. Exceptions which are -raised as a result of DBAPI exceptions are all subclasses of -:exc:`.DBAPIError`. - -""" -from __future__ import annotations - -import typing -from typing import Any -from typing import List -from typing import Optional -from typing import overload -from typing import Tuple -from typing import Type -from typing import Union - -from .util import compat -from .util import preloaded as _preloaded - -if typing.TYPE_CHECKING: - from .engine.interfaces import _AnyExecuteParams - from .engine.interfaces import Dialect - from .sql.compiler import Compiled - from .sql.compiler import TypeCompiler - from .sql.elements import ClauseElement - -if typing.TYPE_CHECKING: - _version_token: str -else: - # set by __init__.py - _version_token = None - - -class HasDescriptionCode: - """helper which adds 'code' as an attribute and '_code_str' as a method""" - - code: Optional[str] = None - - def __init__(self, *arg: Any, **kw: Any): - code = kw.pop("code", None) - if code is not None: - self.code = code - super().__init__(*arg, **kw) - - _what_are_we = "error" - - def _code_str(self) -> str: - if not self.code: - return "" - else: - return ( - f"(Background on this {self._what_are_we} at: " - f"https://sqlalche.me/e/{_version_token}/{self.code})" - ) - - def __str__(self) -> str: - message = super().__str__() - if self.code: - message = "%s %s" % (message, self._code_str()) - return message - - -class SQLAlchemyError(HasDescriptionCode, Exception): - """Generic error class.""" - - def _message(self) -> str: - # rules: - # - # 1. single arg string will usually be a unicode - # object, but since __str__() must return unicode, check for - # bytestring just in case - # - # 2. for multiple self.args, this is not a case in current - # SQLAlchemy though this is happening in at least one known external - # library, call str() which does a repr(). - # - text: str - - if len(self.args) == 1: - arg_text = self.args[0] - - if isinstance(arg_text, bytes): - text = compat.decode_backslashreplace(arg_text, "utf-8") - # This is for when the argument is not a string of any sort. - # Otherwise, converting this exception to string would fail for - # non-string arguments. - else: - text = str(arg_text) - - return text - else: - # this is not a normal case within SQLAlchemy but is here for - # compatibility with Exception.args - the str() comes out as - # a repr() of the tuple - return str(self.args) - - def _sql_message(self) -> str: - message = self._message() - - if self.code: - message = "%s %s" % (message, self._code_str()) - - return message - - def __str__(self) -> str: - return self._sql_message() - - -class ArgumentError(SQLAlchemyError): - """Raised when an invalid or conflicting function argument is supplied. - - This error generally corresponds to construction time state errors. - - """ - - -class DuplicateColumnError(ArgumentError): - """a Column is being added to a Table that would replace another - Column, without appropriate parameters to allow this in place. - - .. versionadded:: 2.0.0b4 - - """ - - -class ObjectNotExecutableError(ArgumentError): - """Raised when an object is passed to .execute() that can't be - executed as SQL. - - """ - - def __init__(self, target: Any): - super().__init__("Not an executable object: %r" % target) - self.target = target - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return self.__class__, (self.target,) - - -class NoSuchModuleError(ArgumentError): - """Raised when a dynamically-loaded module (usually a database dialect) - of a particular name cannot be located.""" - - -class NoForeignKeysError(ArgumentError): - """Raised when no foreign keys can be located between two selectables - during a join.""" - - -class AmbiguousForeignKeysError(ArgumentError): - """Raised when more than one foreign key matching can be located - between two selectables during a join.""" - - -class ConstraintColumnNotFoundError(ArgumentError): - """raised when a constraint refers to a string column name that - is not present in the table being constrained. - - .. versionadded:: 2.0 - - """ - - -class CircularDependencyError(SQLAlchemyError): - """Raised by topological sorts when a circular dependency is detected. - - There are two scenarios where this error occurs: - - * In a Session flush operation, if two objects are mutually dependent - on each other, they can not be inserted or deleted via INSERT or - DELETE statements alone; an UPDATE will be needed to post-associate - or pre-deassociate one of the foreign key constrained values. - The ``post_update`` flag described at :ref:`post_update` can resolve - this cycle. - * In a :attr:`_schema.MetaData.sorted_tables` operation, two - :class:`_schema.ForeignKey` - or :class:`_schema.ForeignKeyConstraint` objects mutually refer to each - other. Apply the ``use_alter=True`` flag to one or both, - see :ref:`use_alter`. - - """ - - def __init__( - self, - message: str, - cycles: Any, - edges: Any, - msg: Optional[str] = None, - code: Optional[str] = None, - ): - if msg is None: - message += " (%s)" % ", ".join(repr(s) for s in cycles) - else: - message = msg - SQLAlchemyError.__init__(self, message, code=code) - self.cycles = cycles - self.edges = edges - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return ( - self.__class__, - (None, self.cycles, self.edges, self.args[0]), - {"code": self.code} if self.code is not None else {}, - ) - - -class CompileError(SQLAlchemyError): - """Raised when an error occurs during SQL compilation""" - - -class UnsupportedCompilationError(CompileError): - """Raised when an operation is not supported by the given compiler. - - .. seealso:: - - :ref:`faq_sql_expression_string` - - :ref:`error_l7de` - """ - - code = "l7de" - - def __init__( - self, - compiler: Union[Compiled, TypeCompiler], - element_type: Type[ClauseElement], - message: Optional[str] = None, - ): - super().__init__( - "Compiler %r can't render element of type %s%s" - % (compiler, element_type, ": %s" % message if message else "") - ) - self.compiler = compiler - self.element_type = element_type - self.message = message - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return self.__class__, (self.compiler, self.element_type, self.message) - - -class IdentifierError(SQLAlchemyError): - """Raised when a schema name is beyond the max character limit""" - - -class DisconnectionError(SQLAlchemyError): - """A disconnect is detected on a raw DB-API connection. - - This error is raised and consumed internally by a connection pool. It can - be raised by the :meth:`_events.PoolEvents.checkout` - event so that the host pool - forces a retry; the exception will be caught three times in a row before - the pool gives up and raises :class:`~sqlalchemy.exc.InvalidRequestError` - regarding the connection attempt. - - """ - - invalidate_pool: bool = False - - -class InvalidatePoolError(DisconnectionError): - """Raised when the connection pool should invalidate all stale connections. - - A subclass of :class:`_exc.DisconnectionError` that indicates that the - disconnect situation encountered on the connection probably means the - entire pool should be invalidated, as the database has been restarted. - - This exception will be handled otherwise the same way as - :class:`_exc.DisconnectionError`, allowing three attempts to reconnect - before giving up. - - .. versionadded:: 1.2 - - """ - - invalidate_pool: bool = True - - -class TimeoutError(SQLAlchemyError): # noqa - """Raised when a connection pool times out on getting a connection.""" - - -class InvalidRequestError(SQLAlchemyError): - """SQLAlchemy was asked to do something it can't do. - - This error generally corresponds to runtime state errors. - - """ - - -class IllegalStateChangeError(InvalidRequestError): - """An object that tracks state encountered an illegal state change - of some kind. - - .. versionadded:: 2.0 - - """ - - -class NoInspectionAvailable(InvalidRequestError): - """A subject passed to :func:`sqlalchemy.inspection.inspect` produced - no context for inspection.""" - - -class PendingRollbackError(InvalidRequestError): - """A transaction has failed and needs to be rolled back before - continuing. - - .. versionadded:: 1.4 - - """ - - -class ResourceClosedError(InvalidRequestError): - """An operation was requested from a connection, cursor, or other - object that's in a closed state.""" - - -class NoSuchColumnError(InvalidRequestError, KeyError): - """A nonexistent column is requested from a ``Row``.""" - - -class NoResultFound(InvalidRequestError): - """A database result was required but none was found. - - - .. versionchanged:: 1.4 This exception is now part of the - ``sqlalchemy.exc`` module in Core, moved from the ORM. The symbol - remains importable from ``sqlalchemy.orm.exc``. - - - """ - - -class MultipleResultsFound(InvalidRequestError): - """A single database result was required but more than one were found. - - .. versionchanged:: 1.4 This exception is now part of the - ``sqlalchemy.exc`` module in Core, moved from the ORM. The symbol - remains importable from ``sqlalchemy.orm.exc``. - - - """ - - -class NoReferenceError(InvalidRequestError): - """Raised by ``ForeignKey`` to indicate a reference cannot be resolved.""" - - table_name: str - - -class AwaitRequired(InvalidRequestError): - """Error raised by the async greenlet spawn if no async operation - was awaited when it required one. - - """ - - code = "xd1r" - - -class MissingGreenlet(InvalidRequestError): - r"""Error raised by the async greenlet await\_ if called while not inside - the greenlet spawn context. - - """ - - code = "xd2s" - - -class NoReferencedTableError(NoReferenceError): - """Raised by ``ForeignKey`` when the referred ``Table`` cannot be - located. - - """ - - def __init__(self, message: str, tname: str): - NoReferenceError.__init__(self, message) - self.table_name = tname - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return self.__class__, (self.args[0], self.table_name) - - -class NoReferencedColumnError(NoReferenceError): - """Raised by ``ForeignKey`` when the referred ``Column`` cannot be - located. - - """ - - def __init__(self, message: str, tname: str, cname: str): - NoReferenceError.__init__(self, message) - self.table_name = tname - self.column_name = cname - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return ( - self.__class__, - (self.args[0], self.table_name, self.column_name), - ) - - -class NoSuchTableError(InvalidRequestError): - """Table does not exist or is not visible to a connection.""" - - -class UnreflectableTableError(InvalidRequestError): - """Table exists but can't be reflected for some reason. - - .. versionadded:: 1.2 - - """ - - -class UnboundExecutionError(InvalidRequestError): - """SQL was attempted without a database connection to execute it on.""" - - -class DontWrapMixin: - """A mixin class which, when applied to a user-defined Exception class, - will not be wrapped inside of :exc:`.StatementError` if the error is - emitted within the process of executing a statement. - - E.g.:: - - from sqlalchemy.exc import DontWrapMixin - - class MyCustomException(Exception, DontWrapMixin): - pass - - class MySpecialType(TypeDecorator): - impl = String - - def process_bind_param(self, value, dialect): - if value == 'invalid': - raise MyCustomException("invalid!") - - """ - - -class StatementError(SQLAlchemyError): - """An error occurred during execution of a SQL statement. - - :class:`StatementError` wraps the exception raised - during execution, and features :attr:`.statement` - and :attr:`.params` attributes which supply context regarding - the specifics of the statement which had an issue. - - The wrapped exception object is available in - the :attr:`.orig` attribute. - - """ - - statement: Optional[str] = None - """The string SQL statement being invoked when this exception occurred.""" - - params: Optional[_AnyExecuteParams] = None - """The parameter list being used when this exception occurred.""" - - orig: Optional[BaseException] = None - """The original exception that was thrown. - - """ - - ismulti: Optional[bool] = None - """multi parameter passed to repr_params(). None is meaningful.""" - - connection_invalidated: bool = False - - def __init__( - self, - message: str, - statement: Optional[str], - params: Optional[_AnyExecuteParams], - orig: Optional[BaseException], - hide_parameters: bool = False, - code: Optional[str] = None, - ismulti: Optional[bool] = None, - ): - SQLAlchemyError.__init__(self, message, code=code) - self.statement = statement - self.params = params - self.orig = orig - self.ismulti = ismulti - self.hide_parameters = hide_parameters - self.detail: List[str] = [] - - def add_detail(self, msg: str) -> None: - self.detail.append(msg) - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return ( - self.__class__, - ( - self.args[0], - self.statement, - self.params, - self.orig, - self.hide_parameters, - self.__dict__.get("code"), - self.ismulti, - ), - {"detail": self.detail}, - ) - - @_preloaded.preload_module("sqlalchemy.sql.util") - def _sql_message(self) -> str: - util = _preloaded.sql_util - - details = [self._message()] - if self.statement: - stmt_detail = "[SQL: %s]" % self.statement - details.append(stmt_detail) - if self.params: - if self.hide_parameters: - details.append( - "[SQL parameters hidden due to hide_parameters=True]" - ) - else: - params_repr = util._repr_params( - self.params, 10, ismulti=self.ismulti - ) - details.append("[parameters: %r]" % params_repr) - code_str = self._code_str() - if code_str: - details.append(code_str) - return "\n".join(["(%s)" % det for det in self.detail] + details) - - -class DBAPIError(StatementError): - """Raised when the execution of a database operation fails. - - Wraps exceptions raised by the DB-API underlying the - database operation. Driver-specific implementations of the standard - DB-API exception types are wrapped by matching sub-types of SQLAlchemy's - :class:`DBAPIError` when possible. DB-API's ``Error`` type maps to - :class:`DBAPIError` in SQLAlchemy, otherwise the names are identical. Note - that there is no guarantee that different DB-API implementations will - raise the same exception type for any given error condition. - - :class:`DBAPIError` features :attr:`~.StatementError.statement` - and :attr:`~.StatementError.params` attributes which supply context - regarding the specifics of the statement which had an issue, for the - typical case when the error was raised within the context of - emitting a SQL statement. - - The wrapped exception object is available in the - :attr:`~.StatementError.orig` attribute. Its type and properties are - DB-API implementation specific. - - """ - - code = "dbapi" - - @overload - @classmethod - def instance( - cls, - statement: Optional[str], - params: Optional[_AnyExecuteParams], - orig: Exception, - dbapi_base_err: Type[Exception], - hide_parameters: bool = False, - connection_invalidated: bool = False, - dialect: Optional[Dialect] = None, - ismulti: Optional[bool] = None, - ) -> StatementError: ... - - @overload - @classmethod - def instance( - cls, - statement: Optional[str], - params: Optional[_AnyExecuteParams], - orig: DontWrapMixin, - dbapi_base_err: Type[Exception], - hide_parameters: bool = False, - connection_invalidated: bool = False, - dialect: Optional[Dialect] = None, - ismulti: Optional[bool] = None, - ) -> DontWrapMixin: ... - - @overload - @classmethod - def instance( - cls, - statement: Optional[str], - params: Optional[_AnyExecuteParams], - orig: BaseException, - dbapi_base_err: Type[Exception], - hide_parameters: bool = False, - connection_invalidated: bool = False, - dialect: Optional[Dialect] = None, - ismulti: Optional[bool] = None, - ) -> BaseException: ... - - @classmethod - def instance( - cls, - statement: Optional[str], - params: Optional[_AnyExecuteParams], - orig: Union[BaseException, DontWrapMixin], - dbapi_base_err: Type[Exception], - hide_parameters: bool = False, - connection_invalidated: bool = False, - dialect: Optional[Dialect] = None, - ismulti: Optional[bool] = None, - ) -> Union[BaseException, DontWrapMixin]: - # Don't ever wrap these, just return them directly as if - # DBAPIError didn't exist. - if ( - isinstance(orig, BaseException) and not isinstance(orig, Exception) - ) or isinstance(orig, DontWrapMixin): - return orig - - if orig is not None: - # not a DBAPI error, statement is present. - # raise a StatementError - if isinstance(orig, SQLAlchemyError) and statement: - return StatementError( - "(%s.%s) %s" - % ( - orig.__class__.__module__, - orig.__class__.__name__, - orig.args[0], - ), - statement, - params, - orig, - hide_parameters=hide_parameters, - code=orig.code, - ismulti=ismulti, - ) - elif not isinstance(orig, dbapi_base_err) and statement: - return StatementError( - "(%s.%s) %s" - % ( - orig.__class__.__module__, - orig.__class__.__name__, - orig, - ), - statement, - params, - orig, - hide_parameters=hide_parameters, - ismulti=ismulti, - ) - - glob = globals() - for super_ in orig.__class__.__mro__: - name = super_.__name__ - if dialect: - name = dialect.dbapi_exception_translation_map.get( - name, name - ) - if name in glob and issubclass(glob[name], DBAPIError): - cls = glob[name] - break - - return cls( - statement, - params, - orig, - connection_invalidated=connection_invalidated, - hide_parameters=hide_parameters, - code=cls.code, - ismulti=ismulti, - ) - - def __reduce__(self) -> Union[str, Tuple[Any, ...]]: - return ( - self.__class__, - ( - self.statement, - self.params, - self.orig, - self.hide_parameters, - self.connection_invalidated, - self.__dict__.get("code"), - self.ismulti, - ), - {"detail": self.detail}, - ) - - def __init__( - self, - statement: Optional[str], - params: Optional[_AnyExecuteParams], - orig: BaseException, - hide_parameters: bool = False, - connection_invalidated: bool = False, - code: Optional[str] = None, - ismulti: Optional[bool] = None, - ): - try: - text = str(orig) - except Exception as e: - text = "Error in str() of DB-API-generated exception: " + str(e) - StatementError.__init__( - self, - "(%s.%s) %s" - % (orig.__class__.__module__, orig.__class__.__name__, text), - statement, - params, - orig, - hide_parameters, - code=code, - ismulti=ismulti, - ) - self.connection_invalidated = connection_invalidated - - -class InterfaceError(DBAPIError): - """Wraps a DB-API InterfaceError.""" - - code = "rvf5" - - -class DatabaseError(DBAPIError): - """Wraps a DB-API DatabaseError.""" - - code = "4xp6" - - -class DataError(DatabaseError): - """Wraps a DB-API DataError.""" - - code = "9h9h" - - -class OperationalError(DatabaseError): - """Wraps a DB-API OperationalError.""" - - code = "e3q8" - - -class IntegrityError(DatabaseError): - """Wraps a DB-API IntegrityError.""" - - code = "gkpj" - - -class InternalError(DatabaseError): - """Wraps a DB-API InternalError.""" - - code = "2j85" - - -class ProgrammingError(DatabaseError): - """Wraps a DB-API ProgrammingError.""" - - code = "f405" - - -class NotSupportedError(DatabaseError): - """Wraps a DB-API NotSupportedError.""" - - code = "tw8g" - - -# Warnings - - -class SATestSuiteWarning(Warning): - """warning for a condition detected during tests that is non-fatal - - Currently outside of SAWarning so that we can work around tools like - Alembic doing the wrong thing with warnings. - - """ - - -class SADeprecationWarning(HasDescriptionCode, DeprecationWarning): - """Issued for usage of deprecated APIs.""" - - deprecated_since: Optional[str] = None - "Indicates the version that started raising this deprecation warning" - - -class Base20DeprecationWarning(SADeprecationWarning): - """Issued for usage of APIs specifically deprecated or legacy in - SQLAlchemy 2.0. - - .. seealso:: - - :ref:`error_b8d9`. - - :ref:`deprecation_20_mode` - - """ - - deprecated_since: Optional[str] = "1.4" - "Indicates the version that started raising this deprecation warning" - - def __str__(self) -> str: - return ( - super().__str__() - + " (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)" - ) - - -class LegacyAPIWarning(Base20DeprecationWarning): - """indicates an API that is in 'legacy' status, a long term deprecation.""" - - -class MovedIn20Warning(Base20DeprecationWarning): - """Subtype of RemovedIn20Warning to indicate an API that moved only.""" - - -class SAPendingDeprecationWarning(PendingDeprecationWarning): - """A similar warning as :class:`_exc.SADeprecationWarning`, this warning - is not used in modern versions of SQLAlchemy. - - """ - - deprecated_since: Optional[str] = None - "Indicates the version that started raising this deprecation warning" - - -class SAWarning(HasDescriptionCode, RuntimeWarning): - """Issued at runtime.""" - - _what_are_we = "warning" |