c34f7862创建于 1月15日历史提交
from textwrap import indent


from mindsdb.api.mysql.mysql_proxy.libs.constants.mysql import ERR


class MindsDBError(Exception):
    pass


class BaseEntityException(MindsDBError):
    """Base exception for entitys errors

    Attributes:
        message (str): error message
        entity_name (str): entity name
    """

    def __init__(self, message: str, entity_name: str = None) -> None:
        self.message = message
        self.entity_name = entity_name or "unknown"

    def __str__(self) -> str:
        return f"{self.message}: {self.entity_name}"


class EntityExistsError(BaseEntityException):
    """Raise when entity exists, but should not"""

    def __init__(self, message: str = None, entity_name: str = None) -> None:
        if message is None:
            message = "Entity exists error"
        super().__init__(message, entity_name)


class EntityNotExistsError(BaseEntityException):
    """Raise when entity not exists, but should"""

    def __init__(self, message: str = None, entity_name: str = None) -> None:
        if message is None:
            message = "Entity does not exists error"
        super().__init__(message, entity_name)


class ParsingError(MindsDBError):
    pass


class QueryError(MindsDBError):
    def __init__(
        self,
        db_name: str | None = None,
        db_type: str | None = None,
        db_error_msg: str | None = None,
        failed_query: str | None = None,
        is_external: bool = True,
        is_expected: bool = False,
    ) -> None:
        self.mysql_error_code = ERR.ER_UNKNOWN_ERROR
        self.db_name = db_name
        self.db_type = db_type
        self.db_error_msg = db_error_msg
        self.failed_query = failed_query
        self.is_external = is_external
        self.is_expected = is_expected

    def __str__(self) -> str:
        return format_db_error_message(
            db_name=self.db_name,
            db_type=self.db_type,
            db_error_msg=self.db_error_msg,
            failed_query=self.failed_query,
            is_external=self.is_external,
        )


def format_db_error_message(
    db_name: str | None = None,
    db_type: str | None = None,
    db_error_msg: str | None = None,
    failed_query: str | None = None,
    is_external: bool = True,
) -> str:
    """Format the error message for the database query.

    Args:
        db_name (str | None): The name of the database.
        db_type (str | None): The type of the database.
        db_error_msg (str | None): The error message.
        failed_query (str | None): The failed query.
        is_external (bool): True if error appeared in external database, False if in internal duckdb

    Returns:
        str: The formatted error message.
    """
    error_message = "Failed to execute external database query during query processing."
    if is_external:
        error_message = (
            "An error occurred while executing a derived query on the external "
            "database during processing of your original SQL query."
        )
    else:
        error_message = (
            "An error occurred while processing an internally generated query derived from your original SQL statement."
        )
    if db_name is not None or db_type is not None:
        error_message += "\n\nDatabase Details:"
        if db_name is not None:
            error_message += f"\n- Name: {db_name}"
        if db_type is not None:
            error_message += f"\n- Type: {db_type}"

    if db_error_msg is not None:
        error_message += f"\n\nError:\n{indent(db_error_msg, '    ')}"

    if failed_query is not None:
        error_message += f"\n\nFailed Query:\n{indent(failed_query, '    ')}"

    return error_message