Skip to content

Errors

The models/errors subpackage defines the custom exception hierarchy for ArchiPy. All errors subclass BaseError and are organized by domain: authentication, authorization, validation, resource management, networking, database, system, Keycloak, and Temporal.

Base Error

The root BaseError class that all ArchiPy exceptions inherit from, providing structured error codes and context.

Classes:

Name Description
BaseError

Base exception class for all custom errors.

BaseError

Base exception class for all custom errors.

This class provides a standardized way to handle errors with support for: - Localization of error messages - Additional context data - Integration with HTTP and gRPC status codes - Template string formatting for dynamic message formatting (using {variable} placeholders) - Text normalization and Persian number conversion

Subclasses should define the following class attributes

code (ClassVar[str]): Error code identifier message_en (ClassVar[str]): English error message (can use {variable} placeholders) message_fa (ClassVar[str]): Persian error message (can use {variable} placeholders) http_status (ClassVar[int]): HTTP status code grpc_status (ClassVar[int]): gRPC status code

Methods:

Name Description
get_message

Gets the localized error message based on the language setting.

to_dict

Converts the exception to a dictionary format for API responses.

abort_grpc_async

Aborts an async gRPC call with the appropriate status code and message.

abort_grpc_sync

Aborts a sync gRPC call with the appropriate status code and message.

abort_with_error_async

Creates an error instance and immediately aborts the async gRPC context.

abort_with_error_sync

Creates an error instance and immediately aborts the sync gRPC context.

Attributes:

Name Type Description
message str

Gets the current language message.

Source code in archipy/models/errors/base_error.py
class BaseError(Exception):
    """Base exception class for all custom errors.

    This class provides a standardized way to handle errors with support for:
    - Localization of error messages
    - Additional context data
    - Integration with HTTP and gRPC status codes
    - Template string formatting for dynamic message formatting (using {variable} placeholders)
    - Text normalization and Persian number conversion

    Subclasses should define the following class attributes:
        code (ClassVar[str]): Error code identifier
        message_en (ClassVar[str]): English error message (can use {variable} placeholders)
        message_fa (ClassVar[str]): Persian error message (can use {variable} placeholders)
        http_status (ClassVar[int]): HTTP status code
        grpc_status (ClassVar[int]): gRPC status code

    """

    # Default error details - subclasses should override these
    code: ClassVar[str] = "UNKNOWN_ERROR"
    message_en: ClassVar[str] = "An unknown error occurred"
    message_fa: ClassVar[str] = "خطای ناشناخته‌ای رخ داده است."
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value
        if HTTP_AVAILABLE and HTTPStatus is not None and HTTPStatus is not None
        else 500
    )
    grpc_status: ClassVar[int] = (
        grpc.StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and grpc is not None and isinstance(grpc.StatusCode.INTERNAL.value, tuple)
        else (grpc.StatusCode.INTERNAL.value if GRPC_AVAILABLE and grpc is not None else 13)
    )

    def __init__(
        self,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
        *args: object,
    ) -> None:
        """Initialize the error with message and optional context.

        Args:
            lang: Language code for the error message (defaults to Persian).
            additional_data: Additional context data for the error.
            *args: Additional arguments for the base Exception class.
        """
        if lang is None:
            try:
                from archipy.configs.base_config import BaseConfig

                self.lang = BaseConfig.global_config().LANGUAGE
            except ImportError, AssertionError:
                self.lang = LanguageType.FA
        else:
            self.lang = lang

        self.additional_data = additional_data or {}

        # Initialize base Exception with the message
        super().__init__(self.get_message(), *args)

    def get_message(self) -> str:
        """Gets the localized error message based on the language setting.

        Returns:
            str: The error message in the current language.
        """
        return self.message_fa if self.lang == LanguageType.FA else self.message_en

    def to_dict(self) -> dict:
        """Converts the exception to a dictionary format for API responses.

        Returns:
            dict: A dictionary containing error details and additional data.
        """
        # Get the processed message (not the template)
        processed_message = self.get_message()

        response = {
            "error": self.code,
            "detail": {
                "code": self.code,
                "message": processed_message,
                "http_status": self.http_status,
                "grpc_status": self.grpc_status,
            },
        }

        # Add additional data if present
        if self.additional_data:
            detail = response["detail"]
            if isinstance(detail, dict):
                detail.update(self.additional_data)

        return response

    def __str__(self) -> str:
        """String representation of the exception.

        Returns:
            str: A formatted string containing the error code and message.
        """
        return (
            f"{self.__class__.__name__}("
            f"code='{self.code}', "
            f"message='{self.get_message()}', "
            f"http_status={self.http_status}, "
            f"grpc_status={self.grpc_status}, "
            f"additional_data={self.additional_data}"
            f")"
        )

    def __repr__(self) -> str:
        """Detailed string representation of the exception.

        Returns:
            str: A detailed string representation including all error details.
        """
        return (
            f"{self.__class__.__name__}("
            f"code='{self.code}', "
            f"message='{self.get_message()}', "
            f"http_status={self.http_status}, "
            f"grpc_status={self.grpc_status}, "
            f"additional_data={self.additional_data}"
            f")"
        )

    @property
    def message(self) -> str:
        """Gets the current language message.

        Returns:
            str: The error message in the current language.
        """
        return self.get_message()

    @staticmethod
    def _convert_int_to_grpc_status(status_int: int) -> grpc.StatusCode:
        """Convert integer status code to gRPC StatusCode enum.

        Args:
            status_int: Integer status code

        Returns:
            grpc.StatusCode: Corresponding StatusCode enum member

        Raises:
            ValueError: If gRPC is not available.
        """
        if not GRPC_AVAILABLE or grpc is None:
            raise ValueError("gRPC is not available")

        status_map = {
            0: grpc.StatusCode.OK,
            1: grpc.StatusCode.CANCELLED,
            2: grpc.StatusCode.UNKNOWN,
            3: grpc.StatusCode.INVALID_ARGUMENT,
            4: grpc.StatusCode.DEADLINE_EXCEEDED,
            5: grpc.StatusCode.NOT_FOUND,
            6: grpc.StatusCode.ALREADY_EXISTS,
            7: grpc.StatusCode.PERMISSION_DENIED,
            8: grpc.StatusCode.RESOURCE_EXHAUSTED,
            9: grpc.StatusCode.FAILED_PRECONDITION,
            10: grpc.StatusCode.ABORTED,
            11: grpc.StatusCode.OUT_OF_RANGE,
            12: grpc.StatusCode.UNIMPLEMENTED,
            13: grpc.StatusCode.INTERNAL,
            14: grpc.StatusCode.UNAVAILABLE,
            15: grpc.StatusCode.DATA_LOSS,
            16: grpc.StatusCode.UNAUTHENTICATED,
        }

        return status_map.get(status_int, grpc.StatusCode.INTERNAL)

    async def abort_grpc_async(self, context: AsyncServicerContext) -> None:
        """Aborts an async gRPC call with the appropriate status code and message.

        Args:
            context: The gRPC ServicerContext to abort.

        Raises:
            ValueError: If context is None or doesn't have abort method.
        """
        if context is None:
            raise ValueError("gRPC context cannot be None")

        if not GRPC_AVAILABLE or not hasattr(context, "abort"):
            raise ValueError("Invalid gRPC context: missing abort method")

        status_code: grpc.StatusCode = self._convert_int_to_grpc_status(self.grpc_status)
        message = self.get_message()

        if self.additional_data and hasattr(context, "set_trailing_metadata"):
            context.set_trailing_metadata((("additional_data", json.dumps(self.additional_data)),))

        if hasattr(context, "abort") and callable(context.abort):
            await context.abort(status_code, message)
        else:
            raise ValueError("gRPC context abort method not available or not callable")

    def abort_grpc_sync(self, context: ServicerContext) -> None:
        """Aborts a sync gRPC call with the appropriate status code and message.

        Args:
            context: The gRPC ServicerContext to abort.

        Raises:
            ValueError: If context is None or doesn't have abort method.
        """
        if context is None:
            raise ValueError("gRPC context cannot be None")

        if not GRPC_AVAILABLE or not hasattr(context, "abort"):
            raise ValueError("Invalid gRPC context: missing abort method")

        status_code: grpc.StatusCode = self._convert_int_to_grpc_status(self.grpc_status)
        message = self.get_message()

        if self.additional_data and hasattr(context, "set_trailing_metadata"):
            context.set_trailing_metadata((("additional_data", json.dumps(self.additional_data)),))

        if hasattr(context, "abort") and callable(context.abort):
            context.abort(status_code, message)
        else:
            raise ValueError("gRPC context abort method not available or not callable")

    @classmethod
    async def abort_with_error_async(
        cls,
        context: AsyncServicerContext,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        """Creates an error instance and immediately aborts the async gRPC context.

        Args:
            context: The async gRPC ServicerContext to abort.
            lang: Language code for the error message.
            additional_data: Additional context data for the error.

        Raises:
            ValueError: If context is None or invalid.
        """
        instance = cls(lang=lang, additional_data=additional_data)
        await instance.abort_grpc_async(context)

    @classmethod
    def abort_with_error_sync(
        cls,
        context: ServicerContext,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        """Creates an error instance and immediately aborts the sync gRPC context.

        Args:
            context: The sync gRPC ServicerContext to abort.
            lang: Language code for the error message.
            additional_data: Additional context data for the error.

        Raises:
            ValueError: If context is None or invalid.
        """
        instance = cls(lang=lang, additional_data=additional_data)
        instance.abort_grpc_sync(context)

message property

message: str

Gets the current language message.

Returns:

Name Type Description
str str

The error message in the current language.

get_message

get_message() -> str

Gets the localized error message based on the language setting.

Returns:

Name Type Description
str str

The error message in the current language.

Source code in archipy/models/errors/base_error.py
def get_message(self) -> str:
    """Gets the localized error message based on the language setting.

    Returns:
        str: The error message in the current language.
    """
    return self.message_fa if self.lang == LanguageType.FA else self.message_en

to_dict

to_dict() -> dict

Converts the exception to a dictionary format for API responses.

Returns:

Name Type Description
dict dict

A dictionary containing error details and additional data.

Source code in archipy/models/errors/base_error.py
def to_dict(self) -> dict:
    """Converts the exception to a dictionary format for API responses.

    Returns:
        dict: A dictionary containing error details and additional data.
    """
    # Get the processed message (not the template)
    processed_message = self.get_message()

    response = {
        "error": self.code,
        "detail": {
            "code": self.code,
            "message": processed_message,
            "http_status": self.http_status,
            "grpc_status": self.grpc_status,
        },
    }

    # Add additional data if present
    if self.additional_data:
        detail = response["detail"]
        if isinstance(detail, dict):
            detail.update(self.additional_data)

    return response

abort_grpc_async async

abort_grpc_async(context: ServicerContext) -> None

Aborts an async gRPC call with the appropriate status code and message.

Parameters:

Name Type Description Default
context ServicerContext

The gRPC ServicerContext to abort.

required

Raises:

Type Description
ValueError

If context is None or doesn't have abort method.

Source code in archipy/models/errors/base_error.py
async def abort_grpc_async(self, context: AsyncServicerContext) -> None:
    """Aborts an async gRPC call with the appropriate status code and message.

    Args:
        context: The gRPC ServicerContext to abort.

    Raises:
        ValueError: If context is None or doesn't have abort method.
    """
    if context is None:
        raise ValueError("gRPC context cannot be None")

    if not GRPC_AVAILABLE or not hasattr(context, "abort"):
        raise ValueError("Invalid gRPC context: missing abort method")

    status_code: grpc.StatusCode = self._convert_int_to_grpc_status(self.grpc_status)
    message = self.get_message()

    if self.additional_data and hasattr(context, "set_trailing_metadata"):
        context.set_trailing_metadata((("additional_data", json.dumps(self.additional_data)),))

    if hasattr(context, "abort") and callable(context.abort):
        await context.abort(status_code, message)
    else:
        raise ValueError("gRPC context abort method not available or not callable")

abort_grpc_sync

abort_grpc_sync(context: ServicerContext) -> None

Aborts a sync gRPC call with the appropriate status code and message.

Parameters:

Name Type Description Default
context ServicerContext

The gRPC ServicerContext to abort.

required

Raises:

Type Description
ValueError

If context is None or doesn't have abort method.

Source code in archipy/models/errors/base_error.py
def abort_grpc_sync(self, context: ServicerContext) -> None:
    """Aborts a sync gRPC call with the appropriate status code and message.

    Args:
        context: The gRPC ServicerContext to abort.

    Raises:
        ValueError: If context is None or doesn't have abort method.
    """
    if context is None:
        raise ValueError("gRPC context cannot be None")

    if not GRPC_AVAILABLE or not hasattr(context, "abort"):
        raise ValueError("Invalid gRPC context: missing abort method")

    status_code: grpc.StatusCode = self._convert_int_to_grpc_status(self.grpc_status)
    message = self.get_message()

    if self.additional_data and hasattr(context, "set_trailing_metadata"):
        context.set_trailing_metadata((("additional_data", json.dumps(self.additional_data)),))

    if hasattr(context, "abort") and callable(context.abort):
        context.abort(status_code, message)
    else:
        raise ValueError("gRPC context abort method not available or not callable")

abort_with_error_async async classmethod

abort_with_error_async(
    context: ServicerContext,
    lang: LanguageType | None = None,
    additional_data: dict[str, Any] | None = None,
) -> None

Creates an error instance and immediately aborts the async gRPC context.

Parameters:

Name Type Description Default
context ServicerContext

The async gRPC ServicerContext to abort.

required
lang LanguageType | None

Language code for the error message.

None
additional_data dict[str, Any] | None

Additional context data for the error.

None

Raises:

Type Description
ValueError

If context is None or invalid.

Source code in archipy/models/errors/base_error.py
@classmethod
async def abort_with_error_async(
    cls,
    context: AsyncServicerContext,
    lang: LanguageType | None = None,
    additional_data: dict[str, Any] | None = None,
) -> None:
    """Creates an error instance and immediately aborts the async gRPC context.

    Args:
        context: The async gRPC ServicerContext to abort.
        lang: Language code for the error message.
        additional_data: Additional context data for the error.

    Raises:
        ValueError: If context is None or invalid.
    """
    instance = cls(lang=lang, additional_data=additional_data)
    await instance.abort_grpc_async(context)

abort_with_error_sync classmethod

abort_with_error_sync(
    context: ServicerContext,
    lang: LanguageType | None = None,
    additional_data: dict[str, Any] | None = None,
) -> None

Creates an error instance and immediately aborts the sync gRPC context.

Parameters:

Name Type Description Default
context ServicerContext

The sync gRPC ServicerContext to abort.

required
lang LanguageType | None

Language code for the error message.

None
additional_data dict[str, Any] | None

Additional context data for the error.

None

Raises:

Type Description
ValueError

If context is None or invalid.

Source code in archipy/models/errors/base_error.py
@classmethod
def abort_with_error_sync(
    cls,
    context: ServicerContext,
    lang: LanguageType | None = None,
    additional_data: dict[str, Any] | None = None,
) -> None:
    """Creates an error instance and immediately aborts the sync gRPC context.

    Args:
        context: The sync gRPC ServicerContext to abort.
        lang: Language code for the error message.
        additional_data: Additional context data for the error.

    Raises:
        ValueError: If context is None or invalid.
    """
    instance = cls(lang=lang, additional_data=additional_data)
    instance.abort_grpc_sync(context)

options: show_root_toc_entry: false heading_level: 3

Auth Errors

Exceptions for authentication and authorization failures.

Classes:

Name Description
UnauthenticatedError

Exception raised when a user is unauthenticated.

InvalidCredentialsError

Exception raised for invalid credentials.

TokenExpiredError

Exception raised when a token has expired.

InvalidTokenError

Exception raised when a token is invalid.

SessionExpiredError

Exception raised when a session has expired.

PermissionDeniedError

Exception raised when permission is denied.

AccountLockedError

Exception raised when an account is locked.

AccountDisabledError

Exception raised when an account is disabled.

InvalidVerificationCodeError

Exception raised when a verification code is invalid.

UnauthenticatedError

Exception raised when a user is unauthenticated.

Source code in archipy/models/errors/auth_errors.py
class UnauthenticatedError(BaseError):
    """Exception raised when a user is unauthenticated."""

    code: ClassVar[str] = "UNAUTHENTICATED"
    message_en: ClassVar[str] = "You are not authorized to perform this action."
    message_fa: ClassVar[str] = "شما مجوز انجام این عمل را ندارید."
    http_status: ClassVar[int] = HTTPStatus.UNAUTHORIZED.value if HTTP_AVAILABLE and HTTPStatus is not None else 401
    grpc_status: ClassVar[int] = (
        StatusCode.UNAUTHENTICATED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAUTHENTICATED.value, tuple)
        else (
            StatusCode.UNAUTHENTICATED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 16
        )
    )

InvalidCredentialsError

Exception raised for invalid credentials.

Methods:

Name Description
get_message

Gets the localized error message with username.

Source code in archipy/models/errors/auth_errors.py
class InvalidCredentialsError(BaseError):
    """Exception raised for invalid credentials."""

    code: ClassVar[str] = "INVALID_CREDENTIALS"
    message_en: ClassVar[str] = "Invalid username or password: {username}"
    message_fa: ClassVar[str] = "نام کاربری یا رمز عبور نامعتبر است: {username}"
    http_status: ClassVar[int] = HTTPStatus.UNAUTHORIZED.value if HTTP_AVAILABLE and HTTPStatus is not None else 401
    grpc_status: ClassVar[int] = (
        StatusCode.UNAUTHENTICATED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAUTHENTICATED.value, tuple)
        else (
            StatusCode.UNAUTHENTICATED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 16
        )
    )

    def __init__(
        self,
        username: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"username": username} if username else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with username."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        username = self.additional_data.get("username", "username")
        return template.format(username=username)

get_message

get_message() -> str

Gets the localized error message with username.

Source code in archipy/models/errors/auth_errors.py
def get_message(self) -> str:
    """Gets the localized error message with username."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    username = self.additional_data.get("username", "username")
    return template.format(username=username)

TokenExpiredError

Exception raised when a token has expired.

Source code in archipy/models/errors/auth_errors.py
class TokenExpiredError(BaseError):
    """Exception raised when a token has expired."""

    code: ClassVar[str] = "TOKEN_EXPIRED"
    message_en: ClassVar[str] = "Authentication token has expired"
    message_fa: ClassVar[str] = "توکن احراز هویت منقضی شده است."
    http_status: ClassVar[int] = HTTPStatus.UNAUTHORIZED.value if HTTP_AVAILABLE and HTTPStatus is not None else 401
    grpc_status: ClassVar[int] = (
        StatusCode.UNAUTHENTICATED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAUTHENTICATED.value, tuple)
        else (
            StatusCode.UNAUTHENTICATED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 16
        )
    )

InvalidTokenError

Exception raised when a token is invalid.

Source code in archipy/models/errors/auth_errors.py
class InvalidTokenError(BaseError):
    """Exception raised when a token is invalid."""

    code: ClassVar[str] = "INVALID_TOKEN"
    message_en: ClassVar[str] = "Invalid authentication token"
    message_fa: ClassVar[str] = "توکن احراز هویت نامعتبر است."
    http_status: ClassVar[int] = HTTPStatus.UNAUTHORIZED.value if HTTP_AVAILABLE and HTTPStatus is not None else 401
    grpc_status: ClassVar[int] = (
        StatusCode.UNAUTHENTICATED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAUTHENTICATED.value, tuple)
        else (
            StatusCode.UNAUTHENTICATED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 16
        )
    )

SessionExpiredError

Exception raised when a session has expired.

Methods:

Name Description
get_message

Gets the localized error message with session ID.

Source code in archipy/models/errors/auth_errors.py
class SessionExpiredError(BaseError):
    """Exception raised when a session has expired."""

    code: ClassVar[str] = "SESSION_EXPIRED"
    message_en: ClassVar[str] = "Session has expired: {session_id}"
    message_fa: ClassVar[str] = "نشست کاربری منقضی شده است: {session_id}"
    http_status: ClassVar[int] = HTTPStatus.UNAUTHORIZED.value if HTTP_AVAILABLE and HTTPStatus is not None else 401
    grpc_status: ClassVar[int] = (
        StatusCode.UNAUTHENTICATED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAUTHENTICATED.value, tuple)
        else (
            StatusCode.UNAUTHENTICATED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 16
        )
    )

    def __init__(
        self,
        session_id: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"session_id": session_id} if session_id else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with session ID."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        session_id = self.additional_data.get("session_id", "session_id")
        return template.format(session_id=session_id)

get_message

get_message() -> str

Gets the localized error message with session ID.

Source code in archipy/models/errors/auth_errors.py
def get_message(self) -> str:
    """Gets the localized error message with session ID."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    session_id = self.additional_data.get("session_id", "session_id")
    return template.format(session_id=session_id)

PermissionDeniedError

Exception raised when permission is denied.

Source code in archipy/models/errors/auth_errors.py
class PermissionDeniedError(BaseError):
    """Exception raised when permission is denied."""

    code: ClassVar[str] = "PERMISSION_DENIED"
    message_en: ClassVar[str] = "Permission denied for this operation"
    message_fa: ClassVar[str] = "دسترسی برای انجام این عملیات وجود ندارد."
    http_status: ClassVar[int] = (
        HTTPStatus.FORBIDDEN.value if HTTP_AVAILABLE and HTTPStatus is not None and HTTPStatus is not None else 403
    )
    grpc_status: ClassVar[int] = (
        StatusCode.PERMISSION_DENIED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.PERMISSION_DENIED.value, tuple)
        else (
            StatusCode.PERMISSION_DENIED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 7
        )
    )

AccountLockedError

Exception raised when an account is locked.

Source code in archipy/models/errors/auth_errors.py
class AccountLockedError(BaseError):
    """Exception raised when an account is locked."""

    code: ClassVar[str] = "ACCOUNT_LOCKED"
    message_en: ClassVar[str] = "Account has been locked due to too many failed attempts"
    message_fa: ClassVar[str] = "حساب کاربری به دلیل تلاش‌های ناموفق متعدد قفل شده است"
    http_status: ClassVar[int] = (
        HTTPStatus.FORBIDDEN.value if HTTP_AVAILABLE and HTTPStatus is not None and HTTPStatus is not None else 403
    )
    grpc_status: ClassVar[int] = (
        StatusCode.PERMISSION_DENIED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.PERMISSION_DENIED.value, tuple)
        else (
            StatusCode.PERMISSION_DENIED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 7
        )
    )

    def __init__(
        self,
        username: str | None = None,
        lockout_duration: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if username:
            data["username"] = username
        if lockout_duration:
            data["lockout_duration"] = lockout_duration
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

AccountDisabledError

Exception raised when an account is disabled.

Source code in archipy/models/errors/auth_errors.py
class AccountDisabledError(BaseError):
    """Exception raised when an account is disabled."""

    code: ClassVar[str] = "ACCOUNT_DISABLED"
    message_en: ClassVar[str] = "Account has been disabled"
    message_fa: ClassVar[str] = "حساب کاربری غیرفعال شده است"
    http_status: ClassVar[int] = (
        HTTPStatus.FORBIDDEN.value if HTTP_AVAILABLE and HTTPStatus is not None and HTTPStatus is not None else 403
    )
    grpc_status: ClassVar[int] = (
        StatusCode.PERMISSION_DENIED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.PERMISSION_DENIED.value, tuple)
        else (
            StatusCode.PERMISSION_DENIED.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 7
        )
    )

    def __init__(
        self,
        username: str | None = None,
        reason: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if username:
            data["username"] = username
        if reason:
            data["reason"] = reason
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidVerificationCodeError

Exception raised when a verification code is invalid.

Source code in archipy/models/errors/auth_errors.py
class InvalidVerificationCodeError(BaseError):
    """Exception raised when a verification code is invalid."""

    code: ClassVar[str] = "INVALID_VERIFICATION_CODE"
    message_en: ClassVar[str] = "Invalid verification code"
    message_fa: ClassVar[str] = "کد تایید نامعتبر است"
    http_status: ClassVar[int] = (
        HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None and HTTPStatus is not None else 400
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (
            StatusCode.INVALID_ARGUMENT.value
            if GRPC_AVAILABLE and StatusCode is not None and StatusCode is not None
            else 3
        )
    )

    def __init__(
        self,
        code: str | None = None,
        remaining_attempts: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if code:
            data["code"] = code
        if remaining_attempts is not None:
            data["remaining_attempts"] = remaining_attempts
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

options: show_root_toc_entry: false heading_level: 3

Validation Errors

Exceptions for input validation and data integrity failures.

Classes:

Name Description
InvalidArgumentError

Exception raised for invalid arguments.

InvalidFormatError

Exception raised for invalid data formats.

InvalidEmailError

Exception raised for invalid email formats.

InvalidPhoneNumberError

Exception raised for invalid phone numbers.

InvalidLandlineNumberError

Exception raised for invalid landline numbers.

InvalidNationalCodeError

Exception raised for invalid national codes.

InvalidPasswordError

Exception raised when a password does not meet the security requirements.

InvalidDateError

Exception raised for invalid date formats.

InvalidUrlError

Exception raised for invalid URL formats.

InvalidIpError

Exception raised for invalid IP address formats.

InvalidJsonError

Exception raised for invalid JSON formats.

InvalidTimestampError

Exception raised when a timestamp format is invalid.

OutOfRangeError

Exception raised when a value is out of range.

InvalidArgumentError

Exception raised for invalid arguments.

Methods:

Name Description
get_message

Gets the localized error message with argument name.

Source code in archipy/models/errors/validation_errors.py
class InvalidArgumentError(BaseError):
    """Exception raised for invalid arguments."""

    code: ClassVar[str] = "INVALID_ARGUMENT"
    message_en: ClassVar[str] = "Invalid argument provided: {argument}"
    message_fa: ClassVar[str] = "پارامتر ورودی نامعتبر است: {argument}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        argument_name: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"argument": argument_name} if argument_name else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with argument name."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        argument = self.additional_data.get("argument", "argument")
        return template.format(argument=argument)

get_message

get_message() -> str

Gets the localized error message with argument name.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with argument name."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    argument = self.additional_data.get("argument", "argument")
    return template.format(argument=argument)

InvalidFormatError

Exception raised for invalid data formats.

Source code in archipy/models/errors/validation_errors.py
class InvalidFormatError(BaseError):
    """Exception raised for invalid data formats."""

    code: ClassVar[str] = "INVALID_FORMAT"
    message_en: ClassVar[str] = "Invalid data format"
    message_fa: ClassVar[str] = "فرمت داده نامعتبر است"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        format_type: str | None = None,
        expected_format: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if format_type:
            data["format_type"] = format_type
        if expected_format:
            data["expected_format"] = expected_format
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidEmailError

Exception raised for invalid email formats.

Methods:

Name Description
get_message

Gets the localized error message with email.

Source code in archipy/models/errors/validation_errors.py
class InvalidEmailError(BaseError):
    """Exception raised for invalid email formats."""

    code: ClassVar[str] = "INVALID_EMAIL"
    message_en: ClassVar[str] = "Invalid email format: {email}"
    message_fa: ClassVar[str] = "فرمت ایمیل نامعتبر است: {email}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        email: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"email": email} if email else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with email."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        email = self.additional_data.get("email", "email")
        return template.format(email=email)

get_message

get_message() -> str

Gets the localized error message with email.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with email."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    email = self.additional_data.get("email", "email")
    return template.format(email=email)

InvalidPhoneNumberError

Exception raised for invalid phone numbers.

Methods:

Name Description
get_message

Gets the localized error message with phone number and normalization.

Source code in archipy/models/errors/validation_errors.py
class InvalidPhoneNumberError(BaseError):
    """Exception raised for invalid phone numbers."""

    code: ClassVar[str] = "INVALID_PHONE"
    message_en: ClassVar[str] = "Invalid Iranian phone number: {phone_number}"
    message_fa: ClassVar[str] = "شماره تلفن همراه ایران نامعتبر است: {phone_number}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        phone_number: str,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"phone_number": phone_number}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data)

    def get_message(self) -> str:
        """Gets the localized error message with phone number and normalization."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        phone_number = self.additional_data.get("phone_number", "phone_number")
        message = template.format(phone_number=phone_number)

        # Convert numbers to Persian if language is FA
        if self.lang == LanguageType.FA:
            message = StringUtils.convert_english_number_to_persian(message)

        return message

get_message

get_message() -> str

Gets the localized error message with phone number and normalization.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with phone number and normalization."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    phone_number = self.additional_data.get("phone_number", "phone_number")
    message = template.format(phone_number=phone_number)

    # Convert numbers to Persian if language is FA
    if self.lang == LanguageType.FA:
        message = StringUtils.convert_english_number_to_persian(message)

    return message

InvalidLandlineNumberError

Exception raised for invalid landline numbers.

Methods:

Name Description
get_message

Gets the localized error message with landline number and normalization.

Source code in archipy/models/errors/validation_errors.py
class InvalidLandlineNumberError(BaseError):
    """Exception raised for invalid landline numbers."""

    code: ClassVar[str] = "INVALID_LANDLINE"
    message_en: ClassVar[str] = "Invalid Iranian landline number: {landline_number}"
    message_fa: ClassVar[str] = "شماره تلفن ثابت ایران نامعتبر است: {landline_number}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        landline_number: str,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"landline_number": landline_number}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data)

    def get_message(self) -> str:
        """Gets the localized error message with landline number and normalization."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        landline_number = self.additional_data.get("landline_number", "landline_number")
        message = template.format(landline_number=landline_number)

        # Convert numbers to Persian if language is FA
        if self.lang == LanguageType.FA:
            message = StringUtils.convert_english_number_to_persian(message)

        return message

get_message

get_message() -> str

Gets the localized error message with landline number and normalization.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with landline number and normalization."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    landline_number = self.additional_data.get("landline_number", "landline_number")
    message = template.format(landline_number=landline_number)

    # Convert numbers to Persian if language is FA
    if self.lang == LanguageType.FA:
        message = StringUtils.convert_english_number_to_persian(message)

    return message

InvalidNationalCodeError

Exception raised for invalid national codes.

Methods:

Name Description
get_message

Gets the localized error message with national code and normalization.

Source code in archipy/models/errors/validation_errors.py
class InvalidNationalCodeError(BaseError):
    """Exception raised for invalid national codes."""

    code: ClassVar[str] = "INVALID_NATIONAL_CODE"
    message_en: ClassVar[str] = "Invalid national code format: {national_code}"
    message_fa: ClassVar[str] = "فرمت کد ملی وارد شده اشتباه است: {national_code}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        national_code: str,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"national_code": national_code}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data)

    def get_message(self) -> str:
        """Gets the localized error message with national code and normalization."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        national_code = self.additional_data.get("national_code", "national_code")
        message = template.format(national_code=national_code)

        # Convert numbers to Persian if language is FA
        if self.lang == LanguageType.FA:
            message = StringUtils.convert_english_number_to_persian(message)

        return message

get_message

get_message() -> str

Gets the localized error message with national code and normalization.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with national code and normalization."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    national_code = self.additional_data.get("national_code", "national_code")
    message = template.format(national_code=national_code)

    # Convert numbers to Persian if language is FA
    if self.lang == LanguageType.FA:
        message = StringUtils.convert_english_number_to_persian(message)

    return message

InvalidPasswordError

Exception raised when a password does not meet the security requirements.

Source code in archipy/models/errors/validation_errors.py
class InvalidPasswordError(BaseError):
    """Exception raised when a password does not meet the security requirements."""

    code: ClassVar[str] = "INVALID_PASSWORD"
    message_en: ClassVar[str] = "Password does not meet the security requirements"
    message_fa: ClassVar[str] = "رمز عبور الزامات امنیتی را برآورده نمی‌کند."
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        requirements: list[str] | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"requirements": requirements} if requirements else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidDateError

Exception raised for invalid date formats.

Source code in archipy/models/errors/validation_errors.py
class InvalidDateError(BaseError):
    """Exception raised for invalid date formats."""

    code: ClassVar[str] = "INVALID_DATE"
    message_en: ClassVar[str] = "Invalid date format"
    message_fa: ClassVar[str] = "فرمت تاریخ نامعتبر است"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        date: str | None = None,
        expected_format: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if date:
            data["date"] = date
        if expected_format:
            data["expected_format"] = expected_format
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidUrlError

Exception raised for invalid URL formats.

Methods:

Name Description
get_message

Gets the localized error message with URL.

Source code in archipy/models/errors/validation_errors.py
class InvalidUrlError(BaseError):
    """Exception raised for invalid URL formats."""

    code: ClassVar[str] = "INVALID_URL"
    message_en: ClassVar[str] = "Invalid URL format: {url}"
    message_fa: ClassVar[str] = "فرمت URL نامعتبر است: {url}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        url: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"url": url} if url else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with URL."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        url = self.additional_data.get("url", "url")
        return template.format(url=url)

get_message

get_message() -> str

Gets the localized error message with URL.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with URL."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    url = self.additional_data.get("url", "url")
    return template.format(url=url)

InvalidIpError

Exception raised for invalid IP address formats.

Methods:

Name Description
get_message

Gets the localized error message with IP address.

Source code in archipy/models/errors/validation_errors.py
class InvalidIpError(BaseError):
    """Exception raised for invalid IP address formats."""

    code: ClassVar[str] = "INVALID_IP"
    message_en: ClassVar[str] = "Invalid IP address format: {ip}"
    message_fa: ClassVar[str] = "فرمت آدرس IP نامعتبر است: {ip}"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        ip: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"ip": ip} if ip else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with IP address."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        ip = self.additional_data.get("ip", "ip")
        return template.format(ip=ip)

get_message

get_message() -> str

Gets the localized error message with IP address.

Source code in archipy/models/errors/validation_errors.py
def get_message(self) -> str:
    """Gets the localized error message with IP address."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    ip = self.additional_data.get("ip", "ip")
    return template.format(ip=ip)

InvalidJsonError

Exception raised for invalid JSON formats.

Source code in archipy/models/errors/validation_errors.py
class InvalidJsonError(BaseError):
    """Exception raised for invalid JSON formats."""

    code: ClassVar[str] = "INVALID_JSON"
    message_en: ClassVar[str] = "Invalid JSON format"
    message_fa: ClassVar[str] = "فرمت JSON نامعتبر است"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        json_data: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if json_data:
            data["json_data"] = json_data
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidTimestampError

Exception raised when a timestamp format is invalid.

Source code in archipy/models/errors/validation_errors.py
class InvalidTimestampError(BaseError):
    """Exception raised when a timestamp format is invalid."""

    code: ClassVar[str] = "INVALID_TIMESTAMP"
    message_en: ClassVar[str] = "Invalid timestamp format"
    message_fa: ClassVar[str] = "فرمت زمان نامعتبر است"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        timestamp: str | None = None,
        expected_format: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if timestamp:
            data["timestamp"] = timestamp
        if expected_format:
            data["expected_format"] = expected_format
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

OutOfRangeError

Exception raised when a value is out of range.

Source code in archipy/models/errors/validation_errors.py
class OutOfRangeError(BaseError):
    """Exception raised when a value is out of range."""

    code: ClassVar[str] = "OUT_OF_RANGE"
    message_en: ClassVar[str] = "Value is out of acceptable range"
    message_fa: ClassVar[str] = "مقدار خارج از محدوده مجاز است."
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.OUT_OF_RANGE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.OUT_OF_RANGE.value, tuple)
        else (StatusCode.OUT_OF_RANGE.value if GRPC_AVAILABLE and StatusCode is not None else 11)
    )

    def __init__(
        self,
        field_name: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"field": field_name} if field_name else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

options: show_root_toc_entry: false heading_level: 3

Resource Errors

Exceptions for resource lifecycle issues such as not found, already exists, and conflict.

Classes:

Name Description
NotFoundError

Exception raised when a resource is not found.

AlreadyExistsError

Exception raised when a resource already exists.

ConflictError

Exception raised when there is a resource conflict.

ResourceLockedError

Exception raised when a resource is locked.

ResourceBusyError

Exception raised when a resource is busy.

DataLossError

Exception raised when data is lost.

InvalidEntityTypeError

Exception raised for invalid entity types.

FileTooLargeError

Exception raised when a file is too large.

InvalidFileTypeError

Exception raised for invalid file types.

QuotaExceededError

Exception raised when a quota is exceeded.

ResourceExhaustedError

Exception raised when a resource is exhausted.

StorageError

Exception raised for storage-related errors.

NotFoundError

Exception raised when a resource is not found.

Methods:

Name Description
get_message

Gets the localized error message with resource type.

Source code in archipy/models/errors/resource_errors.py
class NotFoundError(BaseError):
    """Exception raised when a resource is not found."""

    code: ClassVar[str] = "NOT_FOUND"
    message_en: ClassVar[str] = "Requested resource not found: {resource_type}"
    message_fa: ClassVar[str] = "منبع درخواستی یافت نشد: {resource_type}"
    http_status: ClassVar[int] = HTTPStatus.NOT_FOUND.value if HTTP_AVAILABLE and HTTPStatus is not None else 404
    grpc_status: ClassVar[int] = (
        StatusCode.NOT_FOUND.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.NOT_FOUND.value, tuple)
        else (StatusCode.NOT_FOUND.value if GRPC_AVAILABLE and StatusCode is not None else 5)
    )

    def __init__(
        self,
        resource_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"resource_type": resource_type} if resource_type else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with resource type."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        resource_type = self.additional_data.get("resource_type", "resource_type")
        return template.format(resource_type=resource_type)

get_message

get_message() -> str

Gets the localized error message with resource type.

Source code in archipy/models/errors/resource_errors.py
def get_message(self) -> str:
    """Gets the localized error message with resource type."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    resource_type = self.additional_data.get("resource_type", "resource_type")
    return template.format(resource_type=resource_type)

AlreadyExistsError

Exception raised when a resource already exists.

Methods:

Name Description
get_message

Gets the localized error message with resource type.

Source code in archipy/models/errors/resource_errors.py
class AlreadyExistsError(BaseError):
    """Exception raised when a resource already exists."""

    code: ClassVar[str] = "ALREADY_EXISTS"
    message_en: ClassVar[str] = "Resource already exists: {resource_type}"
    message_fa: ClassVar[str] = "منبع از قبل موجود است: {resource_type}"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ALREADY_EXISTS.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ALREADY_EXISTS.value, tuple)
        else (StatusCode.ALREADY_EXISTS.value if GRPC_AVAILABLE and StatusCode is not None else 6)
    )

    def __init__(
        self,
        resource_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"resource_type": resource_type} if resource_type else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with resource type."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        resource_type = self.additional_data.get("resource_type", "resource_type")
        return template.format(resource_type=resource_type)

get_message

get_message() -> str

Gets the localized error message with resource type.

Source code in archipy/models/errors/resource_errors.py
def get_message(self) -> str:
    """Gets the localized error message with resource type."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    resource_type = self.additional_data.get("resource_type", "resource_type")
    return template.format(resource_type=resource_type)

ConflictError

Exception raised when there is a resource conflict.

Source code in archipy/models/errors/resource_errors.py
class ConflictError(BaseError):
    """Exception raised when there is a resource conflict."""

    code: ClassVar[str] = "CONFLICT"
    message_en: ClassVar[str] = "Resource conflict detected"
    message_fa: ClassVar[str] = "تعارض در منابع تشخیص داده شد"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ABORTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ABORTED.value, tuple)
        else (StatusCode.ABORTED.value if GRPC_AVAILABLE and StatusCode is not None else 10)
    )

    def __init__(
        self,
        resource_type: str | None = None,
        resource_id: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if resource_type:
            data["resource_type"] = resource_type
        if resource_id:
            data["resource_id"] = resource_id
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

ResourceLockedError

Exception raised when a resource is locked.

Source code in archipy/models/errors/resource_errors.py
class ResourceLockedError(BaseError):
    """Exception raised when a resource is locked."""

    code: ClassVar[str] = "RESOURCE_LOCKED"
    message_en: ClassVar[str] = "Resource is currently locked"
    message_fa: ClassVar[str] = "منبع در حال حاضر قفل شده است"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ABORTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ABORTED.value, tuple)
        else (StatusCode.ABORTED.value if GRPC_AVAILABLE and StatusCode is not None else 10)
    )

    def __init__(
        self,
        resource_id: str | None = None,
        lock_owner: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if resource_id:
            data["resource_id"] = resource_id
        if lock_owner:
            data["lock_owner"] = lock_owner
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

ResourceBusyError

Exception raised when a resource is busy.

Source code in archipy/models/errors/resource_errors.py
class ResourceBusyError(BaseError):
    """Exception raised when a resource is busy."""

    code: ClassVar[str] = "RESOURCE_BUSY"
    message_en: ClassVar[str] = "Resource is currently busy"
    message_fa: ClassVar[str] = "منبع در حال حاضر مشغول است"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ABORTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ABORTED.value, tuple)
        else (StatusCode.ABORTED.value if GRPC_AVAILABLE and StatusCode is not None else 10)
    )

    def __init__(
        self,
        resource_id: str | None = None,
        busy_reason: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if resource_id:
            data["resource_id"] = resource_id
        if busy_reason:
            data["busy_reason"] = busy_reason
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

DataLossError

Exception raised when data is lost.

Source code in archipy/models/errors/resource_errors.py
class DataLossError(BaseError):
    """Exception raised when data is lost."""

    code: ClassVar[str] = "DATA_LOSS"
    message_en: ClassVar[str] = "Critical data loss detected"
    message_fa: ClassVar[str] = "از دست دادن اطلاعات حیاتی تشخیص داده شد."
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.DATA_LOSS.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.DATA_LOSS.value, tuple)
        else (StatusCode.DATA_LOSS.value if GRPC_AVAILABLE and StatusCode is not None else 15)
    )

InvalidEntityTypeError

Exception raised for invalid entity types.

Source code in archipy/models/errors/resource_errors.py
class InvalidEntityTypeError(BaseError):
    """Exception raised for invalid entity types."""

    code: ClassVar[str] = "INVALID_ENTITY"
    message_en: ClassVar[str] = "Invalid entity type"
    message_fa: ClassVar[str] = "نوع موجودیت نامعتبر است."
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        message: str | None = None,
        expected_type: str | None = None,
        actual_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if message:
            data["message"] = message
        if expected_type:
            data["expected_type"] = expected_type
        if actual_type:
            data["actual_type"] = actual_type
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

FileTooLargeError

Exception raised when a file is too large.

Source code in archipy/models/errors/resource_errors.py
class FileTooLargeError(BaseError):
    """Exception raised when a file is too large."""

    code: ClassVar[str] = "FILE_TOO_LARGE"
    message_en: ClassVar[str] = "File size exceeds the maximum allowed limit"
    message_fa: ClassVar[str] = "حجم فایل از حد مجاز بیشتر است"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        file_name: str | None = None,
        file_size: int | None = None,
        max_size: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if file_name:
            data["file_name"] = file_name
        if file_size:
            data["file_size"] = file_size
        if max_size:
            data["max_size"] = max_size
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidFileTypeError

Exception raised for invalid file types.

Source code in archipy/models/errors/resource_errors.py
class InvalidFileTypeError(BaseError):
    """Exception raised for invalid file types."""

    code: ClassVar[str] = "INVALID_FILE_TYPE"
    message_en: ClassVar[str] = "File type is not supported"
    message_fa: ClassVar[str] = "نوع فایل پشتیبانی نمی‌شود"
    http_status: ClassVar[int] = HTTPStatus.BAD_REQUEST.value if HTTP_AVAILABLE and HTTPStatus is not None else 400
    grpc_status: ClassVar[int] = (
        StatusCode.INVALID_ARGUMENT.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INVALID_ARGUMENT.value, tuple)
        else (StatusCode.INVALID_ARGUMENT.value if GRPC_AVAILABLE and StatusCode is not None else 3)
    )

    def __init__(
        self,
        file_name: str | None = None,
        file_type: str | None = None,
        allowed_types: list[str] | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if file_name:
            data["file_name"] = file_name
        if file_type:
            data["file_type"] = file_type
        if allowed_types:
            data["allowed_types"] = allowed_types
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

QuotaExceededError

Exception raised when a quota is exceeded.

Source code in archipy/models/errors/resource_errors.py
class QuotaExceededError(BaseError):
    """Exception raised when a quota is exceeded."""

    code: ClassVar[str] = "QUOTA_EXCEEDED"
    message_en: ClassVar[str] = "Storage quota has been exceeded"
    message_fa: ClassVar[str] = "سهمیه ذخیره‌سازی به پایان رسیده است"
    http_status: ClassVar[int] = HTTPStatus.FORBIDDEN.value if HTTP_AVAILABLE and HTTPStatus is not None else 403
    grpc_status: ClassVar[int] = (
        StatusCode.RESOURCE_EXHAUSTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.RESOURCE_EXHAUSTED.value, tuple)
        else (StatusCode.RESOURCE_EXHAUSTED.value if GRPC_AVAILABLE and StatusCode is not None else 8)
    )

    def __init__(
        self,
        quota_type: str | None = None,
        current_usage: int | None = None,
        quota_limit: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if quota_type:
            data["quota_type"] = quota_type
        if current_usage:
            data["current_usage"] = current_usage
        if quota_limit:
            data["quota_limit"] = quota_limit
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

ResourceExhaustedError

Exception raised when a resource is exhausted.

Source code in archipy/models/errors/resource_errors.py
class ResourceExhaustedError(BaseError):
    """Exception raised when a resource is exhausted."""

    code: ClassVar[str] = "RESOURCE_EXHAUSTED"
    message_en: ClassVar[str] = "Resource limit has been reached"
    message_fa: ClassVar[str] = "محدودیت منابع به پایان رسیده است."
    http_status: ClassVar[int] = (
        HTTPStatus.TOO_MANY_REQUESTS.value if HTTP_AVAILABLE and HTTPStatus is not None else 429
    )
    grpc_status: ClassVar[int] = (
        StatusCode.RESOURCE_EXHAUSTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.RESOURCE_EXHAUSTED.value, tuple)
        else (StatusCode.RESOURCE_EXHAUSTED.value if GRPC_AVAILABLE and StatusCode is not None else 8)
    )

    def __init__(
        self,
        resource_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"resource_type": resource_type} if resource_type else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

StorageError

Exception raised for storage-related errors.

Source code in archipy/models/errors/resource_errors.py
class StorageError(BaseError):
    """Exception raised for storage-related errors."""

    code: ClassVar[str] = "STORAGE_ERROR"
    message_en: ClassVar[str] = "Storage access error occurred"
    message_fa: ClassVar[str] = "خطا در دسترسی به فضای ذخیره‌سازی"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        storage_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"storage_type": storage_type} if storage_type else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

options: show_root_toc_entry: false heading_level: 3

Business Errors

Exceptions representing violations of business rules and domain constraints.

Classes:

Name Description
InvalidStateError

Exception raised when an operation is attempted in an invalid state.

FailedPreconditionError

Exception raised when a precondition for an operation is not met.

BusinessRuleViolationError

Exception raised when a business rule is violated.

InvalidOperationError

Exception raised when an operation is not allowed in the current context.

InsufficientFundsError

Exception raised when there are insufficient funds for an operation.

InsufficientBalanceError

Exception raised when an operation fails due to insufficient account balance.

MaintenanceModeError

Exception raised when the system is in maintenance mode.

InvalidStateError

Exception raised when an operation is attempted in an invalid state.

Source code in archipy/models/errors/business_errors.py
class InvalidStateError(BaseError):
    """Exception raised when an operation is attempted in an invalid state."""

    code: ClassVar[str] = "INVALID_STATE"
    message_en: ClassVar[str] = "Invalid state for the requested operation"
    message_fa: ClassVar[str] = "وضعیت نامعتبر برای عملیات درخواستی"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

    def __init__(
        self,
        current_state: str | None = None,
        expected_state: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if current_state:
            data["current_state"] = current_state
        if expected_state:
            data["expected_state"] = expected_state
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

FailedPreconditionError

Exception raised when a precondition for an operation is not met.

Source code in archipy/models/errors/business_errors.py
class FailedPreconditionError(BaseError):
    """Exception raised when a precondition for an operation is not met."""

    code: ClassVar[str] = "FAILED_PRECONDITION"
    message_en: ClassVar[str] = "Operation preconditions not met"
    message_fa: ClassVar[str] = "پیش‌نیازهای عملیات برآورده نشده است."
    http_status: ClassVar[int] = (
        HTTPStatus.PRECONDITION_FAILED.value if HTTP_AVAILABLE and HTTPStatus is not None else 412
    )
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

    def __init__(
        self,
        precondition: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if precondition:
            data["precondition"] = precondition
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

BusinessRuleViolationError

Exception raised when a business rule is violated.

Source code in archipy/models/errors/business_errors.py
class BusinessRuleViolationError(BaseError):
    """Exception raised when a business rule is violated."""

    code: ClassVar[str] = "BUSINESS_RULE_VIOLATION"
    message_en: ClassVar[str] = "Business rule violation"
    message_fa: ClassVar[str] = "نقض قوانین کسب و کار"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

    def __init__(
        self,
        rule: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if rule:
            data["rule"] = rule
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InvalidOperationError

Exception raised when an operation is not allowed in the current context.

Source code in archipy/models/errors/business_errors.py
class InvalidOperationError(BaseError):
    """Exception raised when an operation is not allowed in the current context."""

    code: ClassVar[str] = "INVALID_OPERATION"
    message_en: ClassVar[str] = "Operation is not allowed in the current context"
    message_fa: ClassVar[str] = "عملیات در وضعیت فعلی مجاز نیست"
    http_status: ClassVar[int] = HTTPStatus.FORBIDDEN.value if HTTP_AVAILABLE and HTTPStatus is not None else 403
    grpc_status: ClassVar[int] = (
        StatusCode.PERMISSION_DENIED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.PERMISSION_DENIED.value, tuple)
        else (StatusCode.PERMISSION_DENIED.value if GRPC_AVAILABLE and StatusCode is not None else 7)
    )

    def __init__(
        self,
        operation: str | None = None,
        context: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if operation:
            data["operation"] = operation
        if context:
            data["context"] = context
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

InsufficientFundsError

Exception raised when there are insufficient funds for an operation.

Source code in archipy/models/errors/business_errors.py
class InsufficientFundsError(BaseError):
    """Exception raised when there are insufficient funds for an operation."""

    code: ClassVar[str] = "INSUFFICIENT_FUNDS"
    message_en: ClassVar[str] = "Insufficient funds for the operation"
    message_fa: ClassVar[str] = "موجودی ناکافی برای عملیات"
    http_status: ClassVar[int] = HTTPStatus.PAYMENT_REQUIRED.value if HTTP_AVAILABLE and HTTPStatus is not None else 402
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

InsufficientBalanceError

Exception raised when an operation fails due to insufficient account balance.

Source code in archipy/models/errors/business_errors.py
class InsufficientBalanceError(BaseError):
    """Exception raised when an operation fails due to insufficient account balance."""

    code: ClassVar[str] = "INSUFFICIENT_BALANCE"
    message_en: ClassVar[str] = "Insufficient balance for operation"
    message_fa: ClassVar[str] = "عدم موجودی کافی برای عملیات."
    http_status: ClassVar[int] = HTTPStatus.PAYMENT_REQUIRED.value if HTTP_AVAILABLE and HTTPStatus is not None else 402
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

MaintenanceModeError

Exception raised when the system is in maintenance mode.

Source code in archipy/models/errors/business_errors.py
class MaintenanceModeError(BaseError):
    """Exception raised when the system is in maintenance mode."""

    code: ClassVar[str] = "MAINTENANCE_MODE"
    message_en: ClassVar[str] = "System is currently in maintenance mode"
    message_fa: ClassVar[str] = "سیستم در حال حاضر در حالت تعمیر و نگهداری است"
    http_status: ClassVar[int] = (
        HTTPStatus.SERVICE_UNAVAILABLE.value if HTTP_AVAILABLE and HTTPStatus is not None else 503
    )
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        estimated_duration: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"estimated_duration": estimated_duration} if estimated_duration else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

options: show_root_toc_entry: false heading_level: 3

Network Errors

Exceptions for network communication failures including timeouts and connectivity issues.

Classes:

Name Description
NetworkError

Exception raised for network-related errors.

ConnectionTimeoutError

Exception raised when a connection times out.

ServiceUnavailableError

Exception raised when a service is unavailable.

GatewayTimeoutError

Exception raised when a gateway times out.

BadGatewayError

Exception raised when a gateway returns an invalid response.

RateLimitExceededError

Exception raised when a rate limit is exceeded.

NetworkError

Exception raised for network-related errors.

Source code in archipy/models/errors/network_errors.py
class NetworkError(BaseError):
    """Exception raised for network-related errors."""

    code: ClassVar[str] = "NETWORK_ERROR"
    message_en: ClassVar[str] = "Network error occurred"
    message_fa: ClassVar[str] = "خطای شبکه رخ داده است"
    http_status: ClassVar[int] = HTTPStatus.BAD_GATEWAY.value if HTTP_AVAILABLE and HTTPStatus is not None else 502
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        service: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if service:
            data["service"] = service
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

ConnectionTimeoutError

Exception raised when a connection times out.

Source code in archipy/models/errors/network_errors.py
class ConnectionTimeoutError(BaseError):
    """Exception raised when a connection times out."""

    code: ClassVar[str] = "CONNECTION_TIMEOUT"
    message_en: ClassVar[str] = "Connection timed out"
    message_fa: ClassVar[str] = "اتصال با تایم‌اوت مواجه شد"
    http_status: ClassVar[int] = HTTPStatus.REQUEST_TIMEOUT.value if HTTP_AVAILABLE and HTTPStatus is not None else 408
    grpc_status: ClassVar[int] = (
        StatusCode.DEADLINE_EXCEEDED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.DEADLINE_EXCEEDED.value, tuple)
        else (StatusCode.DEADLINE_EXCEEDED.value if GRPC_AVAILABLE and StatusCode is not None else 4)
    )

    def __init__(
        self,
        service: str | None = None,
        timeout: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if service:
            data["service"] = service
        if timeout:
            data["timeout"] = timeout
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

ServiceUnavailableError

Exception raised when a service is unavailable.

Source code in archipy/models/errors/network_errors.py
class ServiceUnavailableError(BaseError):
    """Exception raised when a service is unavailable."""

    code: ClassVar[str] = "SERVICE_UNAVAILABLE"
    message_en: ClassVar[str] = "Service is currently unavailable"
    message_fa: ClassVar[str] = "سرویس در حال حاضر در دسترس نیست"
    http_status: ClassVar[int] = (
        HTTPStatus.SERVICE_UNAVAILABLE.value if HTTP_AVAILABLE and HTTPStatus is not None else 503
    )
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        service: str | None = None,
        retry_after: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if service:
            data["service"] = service
        if retry_after:
            data["retry_after"] = retry_after
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

GatewayTimeoutError

Exception raised when a gateway times out.

Source code in archipy/models/errors/network_errors.py
class GatewayTimeoutError(BaseError):
    """Exception raised when a gateway times out."""

    code: ClassVar[str] = "GATEWAY_TIMEOUT"
    message_en: ClassVar[str] = "Gateway timeout"
    message_fa: ClassVar[str] = "تایم‌اوت دروازه"
    http_status: ClassVar[int] = HTTPStatus.GATEWAY_TIMEOUT.value if HTTP_AVAILABLE and HTTPStatus is not None else 504
    grpc_status: ClassVar[int] = (
        StatusCode.DEADLINE_EXCEEDED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.DEADLINE_EXCEEDED.value, tuple)
        else (StatusCode.DEADLINE_EXCEEDED.value if GRPC_AVAILABLE and StatusCode is not None else 4)
    )

    def __init__(
        self,
        gateway: str | None = None,
        timeout: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if gateway:
            data["gateway"] = gateway
        if timeout:
            data["timeout"] = timeout
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

BadGatewayError

Exception raised when a gateway returns an invalid response.

Source code in archipy/models/errors/network_errors.py
class BadGatewayError(BaseError):
    """Exception raised when a gateway returns an invalid response."""

    code: ClassVar[str] = "BAD_GATEWAY"
    message_en: ClassVar[str] = "Bad gateway"
    message_fa: ClassVar[str] = "دروازه نامعتبر"
    http_status: ClassVar[int] = HTTPStatus.BAD_GATEWAY.value if HTTP_AVAILABLE and HTTPStatus is not None else 502
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        gateway: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if gateway:
            data["gateway"] = gateway
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

RateLimitExceededError

Exception raised when a rate limit is exceeded.

Source code in archipy/models/errors/network_errors.py
class RateLimitExceededError(BaseError):
    """Exception raised when a rate limit is exceeded."""

    code: ClassVar[str] = "RATE_LIMIT_EXCEEDED"
    message_en: ClassVar[str] = "Rate limit has been exceeded"
    message_fa: ClassVar[str] = "محدودیت نرخ درخواست به پایان رسیده است"
    http_status: ClassVar[int] = (
        HTTPStatus.TOO_MANY_REQUESTS.value if HTTP_AVAILABLE and HTTPStatus is not None else 429
    )
    grpc_status: ClassVar[int] = (
        StatusCode.RESOURCE_EXHAUSTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.RESOURCE_EXHAUSTED.value, tuple)
        else (StatusCode.RESOURCE_EXHAUSTED.value if GRPC_AVAILABLE and StatusCode is not None else 8)
    )

    def __init__(
        self,
        rate_limit_type: str | None = None,
        retry_after: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if rate_limit_type:
            data["rate_limit_type"] = rate_limit_type
        if retry_after:
            data["retry_after"] = retry_after
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

options: show_root_toc_entry: false heading_level: 3

Database Errors

Exceptions for database-level failures including connection errors, constraint violations, and transaction failures.

Classes:

Name Description
DatabaseError

Base class for all database-related errors.

DatabaseConnectionError

Exception raised for database connection errors.

DatabaseQueryError

Exception raised for database query errors.

DatabaseTransactionError

Exception raised for database transaction errors.

DatabaseTimeoutError

Exception raised for database timeout errors.

DatabaseConstraintError

Exception raised for database constraint violations.

DatabaseIntegrityError

Exception raised for database integrity violations.

DatabaseDeadlockError

Exception raised for database deadlock errors.

DatabaseSerializationError

Exception raised for database serialization errors.

DatabaseConfigurationError

Exception raised for database configuration errors.

CacheError

Exception raised for cache access errors.

CacheMissError

Exception raised when requested data is not found in cache.

DatabaseError

Base class for all database-related errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseError(BaseError):
    """Base class for all database-related errors."""

    code: ClassVar[str] = "DATABASE_ERROR"
    message_en: ClassVar[str] = "Database error occurred"
    message_fa: ClassVar[str] = "خطای پایگاه داده رخ داده است"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        database: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if database:
            data["database"] = database
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

DatabaseConnectionError

Exception raised for database connection errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseConnectionError(DatabaseError):
    """Exception raised for database connection errors."""

    code: ClassVar[str] = "DATABASE_CONNECTION_ERROR"
    message_en: ClassVar[str] = "Failed to connect to the database"
    message_fa: ClassVar[str] = "خطا در اتصال به پایگاه داده"
    http_status: ClassVar[int] = (
        HTTPStatus.SERVICE_UNAVAILABLE.value if HTTP_AVAILABLE and HTTPStatus is not None else 503
    )
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

DatabaseQueryError

Exception raised for database query errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseQueryError(DatabaseError):
    """Exception raised for database query errors."""

    code: ClassVar[str] = "DATABASE_QUERY_ERROR"
    message_en: ClassVar[str] = "Error executing database query"
    message_fa: ClassVar[str] = "خطا در اجرای پرس و جوی پایگاه داده"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        database: str | None = None,
        query: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if query:
            data["query"] = query
        if additional_data:
            data.update(additional_data)
        super().__init__(database=database, lang=lang, additional_data=data or None)

DatabaseTransactionError

Exception raised for database transaction errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseTransactionError(DatabaseError):
    """Exception raised for database transaction errors."""

    code: ClassVar[str] = "DATABASE_TRANSACTION_ERROR"
    message_en: ClassVar[str] = "Error in database transaction"
    message_fa: ClassVar[str] = "خطا در تراکنش پایگاه داده"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        database: str | None = None,
        transaction_id: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if transaction_id:
            data["transaction_id"] = transaction_id
        if additional_data:
            data.update(additional_data)
        super().__init__(database=database, lang=lang, additional_data=data or None)

DatabaseTimeoutError

Exception raised for database timeout errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseTimeoutError(DatabaseError):
    """Exception raised for database timeout errors."""

    code: ClassVar[str] = "DATABASE_TIMEOUT_ERROR"
    message_en: ClassVar[str] = "Database operation timed out"
    message_fa: ClassVar[str] = "عملیات پایگاه داده با تایم‌اوت مواجه شد"
    http_status: ClassVar[int] = HTTPStatus.REQUEST_TIMEOUT.value if HTTP_AVAILABLE and HTTPStatus is not None else 408
    grpc_status: ClassVar[int] = (
        StatusCode.DEADLINE_EXCEEDED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.DEADLINE_EXCEEDED.value, tuple)
        else (StatusCode.DEADLINE_EXCEEDED.value if GRPC_AVAILABLE and StatusCode is not None else 4)
    )

    def __init__(
        self,
        database: str | None = None,
        timeout: int | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if timeout:
            data["timeout"] = timeout
        if additional_data:
            data.update(additional_data)
        super().__init__(database=database, lang=lang, additional_data=data or None)

DatabaseConstraintError

Exception raised for database constraint violations.

Source code in archipy/models/errors/database_errors.py
class DatabaseConstraintError(DatabaseError):
    """Exception raised for database constraint violations."""

    code: ClassVar[str] = "DATABASE_CONSTRAINT_ERROR"
    message_en: ClassVar[str] = "Database constraint violation"
    message_fa: ClassVar[str] = "نقض محدودیت پایگاه داده"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

    def __init__(
        self,
        database: str | None = None,
        constraint: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if constraint:
            data["constraint"] = constraint
        if additional_data:
            data.update(additional_data)
        super().__init__(database=database, lang=lang, additional_data=data or None)

DatabaseIntegrityError

Exception raised for database integrity violations.

Source code in archipy/models/errors/database_errors.py
class DatabaseIntegrityError(DatabaseError):
    """Exception raised for database integrity violations."""

    code: ClassVar[str] = "DATABASE_INTEGRITY_ERROR"
    message_en: ClassVar[str] = "Database integrity violation"
    message_fa: ClassVar[str] = "نقض یکپارچگی پایگاه داده"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.FAILED_PRECONDITION.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.FAILED_PRECONDITION.value, tuple)
        else (StatusCode.FAILED_PRECONDITION.value if GRPC_AVAILABLE and StatusCode is not None else 9)
    )

DatabaseDeadlockError

Exception raised for database deadlock errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseDeadlockError(DatabaseError):
    """Exception raised for database deadlock errors."""

    code: ClassVar[str] = "DATABASE_DEADLOCK_ERROR"
    message_en: ClassVar[str] = "Database deadlock detected"
    message_fa: ClassVar[str] = "قفل‌شدگی پایگاه داده تشخیص داده شد"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ABORTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ABORTED.value, tuple)
        else (StatusCode.ABORTED.value if GRPC_AVAILABLE and StatusCode is not None else 10)
    )

DatabaseSerializationError

Exception raised for database serialization errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseSerializationError(DatabaseError):
    """Exception raised for database serialization errors."""

    code: ClassVar[str] = "DATABASE_SERIALIZATION_ERROR"
    message_en: ClassVar[str] = "Database serialization failure"
    message_fa: ClassVar[str] = "خطای سریال‌سازی پایگاه داده"
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ABORTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ABORTED.value, tuple)
        else (StatusCode.ABORTED.value if GRPC_AVAILABLE and StatusCode is not None else 10)
    )

DatabaseConfigurationError

Exception raised for database configuration errors.

Source code in archipy/models/errors/database_errors.py
class DatabaseConfigurationError(DatabaseError):
    """Exception raised for database configuration errors."""

    code: ClassVar[str] = "DATABASE_CONFIGURATION_ERROR"
    message_en: ClassVar[str] = "Database configuration error"
    message_fa: ClassVar[str] = "خطای پیکربندی پایگاه داده"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

CacheError

Exception raised for cache access errors.

Source code in archipy/models/errors/database_errors.py
class CacheError(BaseError):
    """Exception raised for cache access errors."""

    code: ClassVar[str] = "CACHE_ERROR"
    message_en: ClassVar[str] = "Error accessing cache"
    message_fa: ClassVar[str] = "خطا در دسترسی به حافظه نهان"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        cache_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {}
        if cache_type:
            data["cache_type"] = cache_type
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

CacheMissError

Exception raised when requested data is not found in cache.

Methods:

Name Description
get_message

Gets the localized error message with cache key.

Source code in archipy/models/errors/database_errors.py
class CacheMissError(BaseError):
    """Exception raised when requested data is not found in cache."""

    code: ClassVar[str] = "CACHE_MISS"
    message_en: ClassVar[str] = "Requested data not found in cache: {cache_key}"
    message_fa: ClassVar[str] = "داده درخواستی در حافظه نهان یافت نشد: {cache_key}"
    http_status: ClassVar[int] = HTTPStatus.NOT_FOUND.value if HTTP_AVAILABLE and HTTPStatus is not None else 404
    grpc_status: ClassVar[int] = (
        StatusCode.NOT_FOUND.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.NOT_FOUND.value, tuple)
        else (StatusCode.NOT_FOUND.value if GRPC_AVAILABLE and StatusCode is not None else 5)
    )

    def __init__(
        self,
        cache_key: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict | None = None,
    ) -> None:
        data = {"cache_key": cache_key} if cache_key else {}
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

    def get_message(self) -> str:
        """Gets the localized error message with cache key."""
        template = self.message_fa if self.lang == LanguageType.FA else self.message_en
        cache_key = self.additional_data.get("cache_key", "cache_key")
        return template.format(cache_key=cache_key)

get_message

get_message() -> str

Gets the localized error message with cache key.

Source code in archipy/models/errors/database_errors.py
def get_message(self) -> str:
    """Gets the localized error message with cache key."""
    template = self.message_fa if self.lang == LanguageType.FA else self.message_en
    cache_key = self.additional_data.get("cache_key", "cache_key")
    return template.format(cache_key=cache_key)

options: show_root_toc_entry: false heading_level: 3

System Errors

Exceptions for system-level failures including configuration errors and unexpected runtime conditions.

Classes:

Name Description
InternalError

Represents an internal server error.

ConfigurationError

Represents a configuration error.

UnavailableError

Represents a resource unavailability error.

UnknownError

Represents an unknown error.

AbortedError

Represents an aborted operation error.

DeadlockDetectedError

Represents a deadlock detection error.

DeadlineExceededError

Raised when an operation exceeds its deadline/timeout.

DeprecationError

Raised when deprecated functionality is used.

InternalError

Represents an internal server error.

This error is typically used when an unexpected condition is encountered that prevents the server from fulfilling the request.

Source code in archipy/models/errors/system_errors.py
class InternalError(BaseError):
    """Represents an internal server error.

    This error is typically used when an unexpected condition is encountered
    that prevents the server from fulfilling the request.
    """

    code: ClassVar[str] = "INTERNAL_ERROR"
    message_en: ClassVar[str] = "Internal system error occurred"
    message_fa: ClassVar[str] = "خطای داخلی سیستم رخ داده است."
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        error_code: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        data = {}
        if error_code:
            data["error_code"] = error_code
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

ConfigurationError

Represents a configuration error.

This error is used when there is a problem with the application's configuration that prevents it from operating correctly.

Source code in archipy/models/errors/system_errors.py
class ConfigurationError(BaseError):
    """Represents a configuration error.

    This error is used when there is a problem with the application's
    configuration that prevents it from operating correctly.
    """

    code: ClassVar[str] = "CONFIGURATION_ERROR"
    message_en: ClassVar[str] = "Error in system configuration"
    message_fa: ClassVar[str] = "خطا در پیکربندی سیستم"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        operation: str | None = None,
        reason: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        data = {}
        if operation:
            data["operation"] = operation
        if reason:
            data["reason"] = reason
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

UnavailableError

Represents a resource unavailability error.

This error is used when a required resource is temporarily unavailable but may become available again in the future.

Source code in archipy/models/errors/system_errors.py
class UnavailableError(BaseError):
    """Represents a resource unavailability error.

    This error is used when a required resource is temporarily unavailable
    but may become available again in the future.
    """

    code: ClassVar[str] = "UNAVAILABLE"
    message_en: ClassVar[str] = "Service is currently unavailable"
    message_fa: ClassVar[str] = "سرویس در حال حاضر در دسترس نیست."
    http_status: ClassVar[int] = (
        HTTPStatus.SERVICE_UNAVAILABLE.value if HTTP_AVAILABLE and HTTPStatus is not None else 503
    )
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        resource_type: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        data = {}
        if resource_type:
            data["resource_type"] = resource_type
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

UnknownError

Represents an unknown error.

This is a catch-all error type for unexpected conditions that don't fit into other error categories.

Source code in archipy/models/errors/system_errors.py
class UnknownError(BaseError):
    """Represents an unknown error.

    This is a catch-all error type for unexpected conditions that
    don't fit into other error categories.
    """

    code: ClassVar[str] = "UNKNOWN_ERROR"
    message_en: ClassVar[str] = "An unknown error occurred"
    message_fa: ClassVar[str] = "خطای ناشناخته‌ای رخ داده است."
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.UNKNOWN.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNKNOWN.value, tuple)
        else (StatusCode.UNKNOWN.value if GRPC_AVAILABLE and StatusCode is not None else 2)
    )

    def __init__(
        self,
        config_key: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        data = {}
        if config_key:
            data["config_key"] = config_key
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

AbortedError

Represents an aborted operation error.

This error is used when an operation is aborted, typically due to a concurrency issue or user cancellation.

Source code in archipy/models/errors/system_errors.py
class AbortedError(BaseError):
    """Represents an aborted operation error.

    This error is used when an operation is aborted, typically due to
    a concurrency issue or user cancellation.
    """

    code: ClassVar[str] = "ABORTED"
    message_en: ClassVar[str] = "Operation was aborted"
    message_fa: ClassVar[str] = "عملیات متوقف شد."
    http_status: ClassVar[int] = HTTPStatus.CONFLICT.value if HTTP_AVAILABLE and HTTPStatus is not None else 409
    grpc_status: ClassVar[int] = (
        StatusCode.ABORTED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.ABORTED.value, tuple)
        else (StatusCode.ABORTED.value if GRPC_AVAILABLE and StatusCode is not None else 10)
    )

    def __init__(
        self,
        service: str | None = None,
        reason: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        data = {}
        if service:
            data["service"] = service
        if reason:
            data["reason"] = reason
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

DeadlockDetectedError

Represents a deadlock detection error.

This error is used when a deadlock is detected in a system operation, typically in database transactions or resource locking scenarios.

Source code in archipy/models/errors/system_errors.py
class DeadlockDetectedError(BaseError):
    """Represents a deadlock detection error.

    This error is used when a deadlock is detected in a system operation,
    typically in database transactions or resource locking scenarios.
    """

    code: ClassVar[str] = "DEADLOCK"
    message_en: ClassVar[str] = "Deadlock detected"
    message_fa: ClassVar[str] = "خطای قفل‌شدگی (Deadlock) تشخیص داده شد."
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        service: str | None = None,
        reason: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        data = {}
        if service:
            data["service"] = service
        if reason:
            data["reason"] = reason
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

DeadlineExceededError

Raised when an operation exceeds its deadline/timeout.

This error is typically used in decorators or functions that have time limits or deadlines for completion.

Source code in archipy/models/errors/system_errors.py
class DeadlineExceededError(BaseError):
    """Raised when an operation exceeds its deadline/timeout.

    This error is typically used in decorators or functions that have
    time limits or deadlines for completion.
    """

    code: ClassVar[str] = "DEADLINE_EXCEEDED"
    message_en: ClassVar[str] = "Operation exceeded its deadline"
    message_fa: ClassVar[str] = "عملیات از مهلت زمانی مجاز تجاوز کرد"
    http_status: ClassVar[int] = HTTPStatus.REQUEST_TIMEOUT.value if HTTP_AVAILABLE and HTTPStatus is not None else 408
    grpc_status: ClassVar[int] = (
        StatusCode.DEADLINE_EXCEEDED.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.DEADLINE_EXCEEDED.value, tuple)
        else (StatusCode.DEADLINE_EXCEEDED.value if GRPC_AVAILABLE and StatusCode is not None else 4)
    )

    def __init__(
        self,
        timeout: int | None = None,
        operation: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        """Initialize DeadlineExceededError.

        Args:
            timeout: The timeout value that was exceeded (in seconds).
            operation: The operation that exceeded the deadline.
            lang: The language for error messages.
            additional_data: Additional context data.
        """
        data = {}
        if timeout is not None:
            data["timeout"] = timeout
        if operation:
            data["operation"] = operation
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

DeprecationError

Raised when deprecated functionality is used.

This error is used to signal that a feature, method, or API is deprecated and should no longer be used.

Source code in archipy/models/errors/system_errors.py
class DeprecationError(BaseError):
    """Raised when deprecated functionality is used.

    This error is used to signal that a feature, method, or API
    is deprecated and should no longer be used.
    """

    code: ClassVar[str] = "DEPRECATED_FEATURE"
    message_en: ClassVar[str] = "This feature is deprecated and should no longer be used"
    message_fa: ClassVar[str] = "این ویژگی منسوخ شده و دیگر نباید استفاده شود"
    http_status: ClassVar[int] = HTTPStatus.GONE.value if HTTP_AVAILABLE and HTTPStatus is not None else 410
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        deprecated_feature: str | None = None,
        replacement: str | None = None,
        removal_version: str | None = None,
        lang: LanguageType | None = None,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        """Initialize DeprecationError.

        Args:
            deprecated_feature: The name of the deprecated feature.
            replacement: The recommended replacement feature.
            removal_version: The version when the feature will be removed.
            lang: The language for error messages.
            additional_data: Additional context data.
        """
        data = {}
        if deprecated_feature:
            data["deprecated_feature"] = deprecated_feature
        if replacement:
            data["replacement"] = replacement
        if removal_version:
            data["removal_version"] = removal_version
        if additional_data:
            data.update(additional_data)
        super().__init__(lang=lang, additional_data=data or None)

options: show_root_toc_entry: false heading_level: 3

Keycloak Errors

Exceptions specific to Keycloak integration failures, such as token validation errors and realm configuration issues.

Classes:

Name Description
RealmAlreadyExistsError

Exception raised when trying to create a realm that already exists.

UserAlreadyExistsError

Exception raised when trying to create a user that already exists.

ClientAlreadyExistsError

Exception raised when trying to create a client that already exists.

RoleAlreadyExistsError

Exception raised when trying to create a role that already exists.

InvalidCredentialsError

Exception raised for invalid authentication credentials.

ResourceNotFoundError

Exception raised when a resource is not found.

InsufficientPermissionsError

Exception raised when user lacks required permissions.

ValidationError

Exception raised for validation errors.

PasswordPolicyError

Exception raised when password doesn't meet policy requirements.

KeycloakConnectionTimeoutError

Exception raised when Keycloak connection times out.

KeycloakServiceUnavailableError

Exception raised when Keycloak service is unavailable.

Functions:

Name Description
get_error_message

Extract the actual error message from Keycloak error.

handle_keycloak_error

Convert Keycloak error to appropriate custom error.

RealmAlreadyExistsError

Exception raised when trying to create a realm that already exists.

Source code in archipy/models/errors/keycloak_errors.py
class RealmAlreadyExistsError(BaseError):
    """Exception raised when trying to create a realm that already exists."""

    code: ClassVar[str] = "REALM_ALREADY_EXISTS"
    message_en: ClassVar[str] = "Realm already exists"
    message_fa: ClassVar[str] = "قلمرو از قبل وجود دارد"
    http_status: ClassVar[int] = 409
    grpc_status: ClassVar[int] = 6

UserAlreadyExistsError

Exception raised when trying to create a user that already exists.

Source code in archipy/models/errors/keycloak_errors.py
class UserAlreadyExistsError(BaseError):
    """Exception raised when trying to create a user that already exists."""

    code: ClassVar[str] = "USER_ALREADY_EXISTS"
    message_en: ClassVar[str] = "User already exists"
    message_fa: ClassVar[str] = "کاربر از قبل وجود دارد"
    http_status: ClassVar[int] = 409
    grpc_status: ClassVar[int] = 6

ClientAlreadyExistsError

Exception raised when trying to create a client that already exists.

Source code in archipy/models/errors/keycloak_errors.py
class ClientAlreadyExistsError(BaseError):
    """Exception raised when trying to create a client that already exists."""

    code: ClassVar[str] = "CLIENT_ALREADY_EXISTS"
    message_en: ClassVar[str] = "Client already exists"
    message_fa: ClassVar[str] = "کلاینت از قبل وجود دارد"
    http_status: ClassVar[int] = 409
    grpc_status: ClassVar[int] = 6

RoleAlreadyExistsError

Exception raised when trying to create a role that already exists.

Source code in archipy/models/errors/keycloak_errors.py
class RoleAlreadyExistsError(BaseError):
    """Exception raised when trying to create a role that already exists."""

    code: ClassVar[str] = "ROLE_ALREADY_EXISTS"
    message_en: ClassVar[str] = "Role already exists"
    message_fa: ClassVar[str] = "نقش از قبل وجود دارد"
    http_status: ClassVar[int] = 409
    grpc_status: ClassVar[int] = 6

InvalidCredentialsError

Exception raised for invalid authentication credentials.

Source code in archipy/models/errors/keycloak_errors.py
class InvalidCredentialsError(BaseError):
    """Exception raised for invalid authentication credentials."""

    code: ClassVar[str] = "INVALID_CREDENTIALS"
    message_en: ClassVar[str] = "Invalid credentials"
    message_fa: ClassVar[str] = "اطلاعات ورود نامعتبر"
    http_status: ClassVar[int] = 401
    grpc_status: ClassVar[int] = 16

ResourceNotFoundError

Exception raised when a resource is not found.

Source code in archipy/models/errors/keycloak_errors.py
class ResourceNotFoundError(BaseError):
    """Exception raised when a resource is not found."""

    code: ClassVar[str] = "RESOURCE_NOT_FOUND"
    message_en: ClassVar[str] = "Resource not found"
    message_fa: ClassVar[str] = "منبع یافت نشد"
    http_status: ClassVar[int] = 404
    grpc_status: ClassVar[int] = 5

InsufficientPermissionsError

Exception raised when user lacks required permissions.

Source code in archipy/models/errors/keycloak_errors.py
class InsufficientPermissionsError(BaseError):
    """Exception raised when user lacks required permissions."""

    code: ClassVar[str] = "INSUFFICIENT_PERMISSIONS"
    message_en: ClassVar[str] = "Insufficient permissions"
    message_fa: ClassVar[str] = "دسترسی کافی نیست"
    http_status: ClassVar[int] = 403
    grpc_status: ClassVar[int] = 7

ValidationError

Exception raised for validation errors.

Source code in archipy/models/errors/keycloak_errors.py
class ValidationError(BaseError):
    """Exception raised for validation errors."""

    code: ClassVar[str] = "VALIDATION_ERROR"
    message_en: ClassVar[str] = "Validation error"
    message_fa: ClassVar[str] = "خطای اعتبارسنجی"
    http_status: ClassVar[int] = 400
    grpc_status: ClassVar[int] = 3

PasswordPolicyError

Exception raised when password doesn't meet policy requirements.

Source code in archipy/models/errors/keycloak_errors.py
class PasswordPolicyError(BaseError):
    """Exception raised when password doesn't meet policy requirements."""

    code: ClassVar[str] = "PASSWORD_POLICY_VIOLATION"
    message_en: ClassVar[str] = "Password does not meet policy requirements"
    message_fa: ClassVar[str] = "رمز عبور الزامات سیاست را برآورده نمی‌کند"
    http_status: ClassVar[int] = 400
    grpc_status: ClassVar[int] = 3

KeycloakConnectionTimeoutError

Exception raised when Keycloak connection times out.

Source code in archipy/models/errors/keycloak_errors.py
class KeycloakConnectionTimeoutError(BaseError):
    """Exception raised when Keycloak connection times out."""

    code: ClassVar[str] = "CONNECTION_TIMEOUT"
    message_en: ClassVar[str] = "Connection timeout"
    message_fa: ClassVar[str] = "زمان اتصال به پایان رسید"
    http_status: ClassVar[int] = 504
    grpc_status: ClassVar[int] = 4

KeycloakServiceUnavailableError

Exception raised when Keycloak service is unavailable.

Source code in archipy/models/errors/keycloak_errors.py
class KeycloakServiceUnavailableError(BaseError):
    """Exception raised when Keycloak service is unavailable."""

    code: ClassVar[str] = "SERVICE_UNAVAILABLE"
    message_en: ClassVar[str] = "Service unavailable"
    message_fa: ClassVar[str] = "سرویس در دسترس نیست"
    http_status: ClassVar[int] = 503
    grpc_status: ClassVar[int] = 14

get_error_message

get_error_message(keycloak_error: KeycloakError) -> str

Extract the actual error message from Keycloak error.

Source code in archipy/models/errors/keycloak_errors.py
def get_error_message(keycloak_error: KeycloakError) -> str:
    """Extract the actual error message from Keycloak error."""
    error_message = str(keycloak_error)

    # Try to parse JSON response body
    if hasattr(keycloak_error, "response_body") and keycloak_error.response_body:
        try:
            body = keycloak_error.response_body
            body_str = body.decode("utf-8") if isinstance(body, bytes) else str(body)

            # body_str is now guaranteed to be str after decode
            parsed = json.loads(body_str)
            if isinstance(parsed, dict):
                error_message = (
                    parsed.get("errorMessage")
                    or parsed.get("error_description")
                    or parsed.get("error")
                    or error_message
                )
        except json.JSONDecodeError, UnicodeDecodeError:
            pass

    return error_message

handle_keycloak_error

handle_keycloak_error(
    keycloak_error: KeycloakError, **additional_data: Any
) -> BaseError

Convert Keycloak error to appropriate custom error.

Source code in archipy/models/errors/keycloak_errors.py
def handle_keycloak_error(keycloak_error: KeycloakError, **additional_data: Any) -> BaseError:
    """Convert Keycloak error to appropriate custom error."""
    error_message = get_error_message(keycloak_error)
    response_code = getattr(keycloak_error, "response_code", None)

    # Add context data
    context = {
        "original_error": error_message,
        "response_code": response_code,
        "keycloak_error_type": type(keycloak_error).__name__,
        **additional_data,
    }

    # Simple string matching to identify error types
    error_lower = error_message.lower()

    # Realm errors
    if "realm" in error_lower and "already exists" in error_lower:
        return RealmAlreadyExistsError(additional_data=context)

    # User errors
    if "user exists with same" in error_lower:
        return UserAlreadyExistsError(additional_data=context)

    # Client errors
    if "client" in error_lower and "already exists" in error_lower:
        return ClientAlreadyExistsError(additional_data=context)

    # Authentication errors
    if any(
        phrase in error_lower for phrase in ["invalid user credentials", "invalid credentials", "authentication failed"]
    ):
        return InvalidCredentialsError(additional_data=context)

    # Not found errors
    if "not found" in error_lower:
        return ResourceNotFoundError(additional_data=context)

    # Permission errors
    if any(phrase in error_lower for phrase in ["forbidden", "access denied", "insufficient permissions"]):
        return InsufficientPermissionsError(additional_data=context)

    # Validation errors (400 status codes that don't match above)
    if response_code == 400:
        return ValidationError(additional_data=context)

    # Default to InternalError for unrecognized errors
    return InternalError(additional_data=context)

options: show_root_toc_entry: false heading_level: 3

Temporal Errors

Exceptions specific to Temporal workflow orchestration failures, such as workflow execution errors and activity timeouts.

Temporal-specific error definitions.

This module defines custom exception classes for Temporal worker operations, extending the base ArchiPy error handling patterns.

Classes:

Name Description
TemporalError

Base exception for all Temporal-related errors.

WorkerConnectionError

Exception raised when a worker fails to connect to Temporal server.

WorkerShutdownError

Exception raised when a worker fails to shutdown gracefully.

TemporalError

Base exception for all Temporal-related errors.

This is the root exception class for all Temporal workflow engine errors within the ArchiPy system.

Source code in archipy/models/errors/temporal_errors.py
class TemporalError(BaseError):
    """Base exception for all Temporal-related errors.

    This is the root exception class for all Temporal workflow engine errors
    within the ArchiPy system.
    """

    code: ClassVar[str] = "TEMPORAL_ERROR"
    message_en: ClassVar[str] = "Temporal error occurred"
    message_fa: ClassVar[str] = "خطای Temporal رخ داده است"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

WorkerConnectionError

Exception raised when a worker fails to connect to Temporal server.

Source code in archipy/models/errors/temporal_errors.py
class WorkerConnectionError(TemporalError):
    """Exception raised when a worker fails to connect to Temporal server."""

    code: ClassVar[str] = "WORKER_CONNECTION_ERROR"
    message_en: ClassVar[str] = "Failed to connect to Temporal server"
    message_fa: ClassVar[str] = "خطا در اتصال به سرور Temporal"
    http_status: ClassVar[int] = (
        HTTPStatus.SERVICE_UNAVAILABLE.value if HTTP_AVAILABLE and HTTPStatus is not None else 503
    )
    grpc_status: ClassVar[int] = (
        StatusCode.UNAVAILABLE.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.UNAVAILABLE.value, tuple)
        else (StatusCode.UNAVAILABLE.value if GRPC_AVAILABLE and StatusCode is not None else 14)
    )

    def __init__(
        self,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        """Initialize the worker connection error."""
        super().__init__(additional_data=additional_data)

WorkerShutdownError

Exception raised when a worker fails to shutdown gracefully.

Source code in archipy/models/errors/temporal_errors.py
class WorkerShutdownError(TemporalError):
    """Exception raised when a worker fails to shutdown gracefully."""

    code: ClassVar[str] = "WORKER_SHUTDOWN_ERROR"
    message_en: ClassVar[str] = "Failed to shutdown Temporal worker gracefully"
    message_fa: ClassVar[str] = "خطا در خاموش‌سازی صحیح کارگر Temporal"
    http_status: ClassVar[int] = (
        HTTPStatus.INTERNAL_SERVER_ERROR.value if HTTP_AVAILABLE and HTTPStatus is not None else 500
    )
    grpc_status: ClassVar[int] = (
        StatusCode.INTERNAL.value[0]
        if GRPC_AVAILABLE and StatusCode is not None and isinstance(StatusCode.INTERNAL.value, tuple)
        else (StatusCode.INTERNAL.value if GRPC_AVAILABLE and StatusCode is not None else 13)
    )

    def __init__(
        self,
        additional_data: dict[str, Any] | None = None,
    ) -> None:
        """Initialize the worker shutdown error."""
        super().__init__(additional_data=additional_data)

options: show_root_toc_entry: false heading_level: 3