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.
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
classBaseError(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 thesecode:ClassVar[str]="UNKNOWN_ERROR"message_en:ClassVar[str]="An unknown error occurred"message_fa:ClassVar[str]="خطای ناشناختهای رخ داده است."http_status:ClassVar[int]=(HTTPStatus.INTERNAL_SERVER_ERROR.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(grpc.StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandgrpcisnotNoneandisinstance(grpc.StatusCode.INTERNAL.value,tuple)else(grpc.StatusCode.INTERNAL.valueifGRPC_AVAILABLEandgrpcisnotNoneelse13))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. """iflangisNone:try:fromarchipy.configs.base_configimportBaseConfigself.lang=BaseConfig.global_config().LANGUAGEexceptImportError,AssertionError:self.lang=LanguageType.FAelse:self.lang=langself.additional_data=additional_dataor{}# Initialize base Exception with the messagesuper().__init__(self.get_message(),*args)defget_message(self)->str:"""Gets the localized error message based on the language setting. Returns: str: The error message in the current language. """returnself.message_faifself.lang==LanguageType.FAelseself.message_endefto_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 presentifself.additional_data:detail=response["detail"]ifisinstance(detail,dict):detail.update(self.additional_data)returnresponsedef__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")")@propertydefmessage(self)->str:"""Gets the current language message. Returns: str: The error message in the current language. """returnself.get_message()@staticmethoddef_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. """ifnotGRPC_AVAILABLEorgrpcisNone:raiseValueError("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,}returnstatus_map.get(status_int,grpc.StatusCode.INTERNAL)asyncdefabort_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. """ifcontextisNone:raiseValueError("gRPC context cannot be None")ifnotGRPC_AVAILABLEornothasattr(context,"abort"):raiseValueError("Invalid gRPC context: missing abort method")status_code:grpc.StatusCode=self._convert_int_to_grpc_status(self.grpc_status)message=self.get_message()ifself.additional_dataandhasattr(context,"set_trailing_metadata"):context.set_trailing_metadata((("additional_data",json.dumps(self.additional_data)),))ifhasattr(context,"abort")andcallable(context.abort):awaitcontext.abort(status_code,message)else:raiseValueError("gRPC context abort method not available or not callable")defabort_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. """ifcontextisNone:raiseValueError("gRPC context cannot be None")ifnotGRPC_AVAILABLEornothasattr(context,"abort"):raiseValueError("Invalid gRPC context: missing abort method")status_code:grpc.StatusCode=self._convert_int_to_grpc_status(self.grpc_status)message=self.get_message()ifself.additional_dataandhasattr(context,"set_trailing_metadata"):context.set_trailing_metadata((("additional_data",json.dumps(self.additional_data)),))ifhasattr(context,"abort")andcallable(context.abort):context.abort(status_code,message)else:raiseValueError("gRPC context abort method not available or not callable")@classmethodasyncdefabort_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)awaitinstance.abort_grpc_async(context)@classmethoddefabort_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)
defget_message(self)->str:"""Gets the localized error message based on the language setting. Returns: str: The error message in the current language. """returnself.message_faifself.lang==LanguageType.FAelseself.message_en
defto_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 presentifself.additional_data:detail=response["detail"]ifisinstance(detail,dict):detail.update(self.additional_data)returnresponse
asyncdefabort_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. """ifcontextisNone:raiseValueError("gRPC context cannot be None")ifnotGRPC_AVAILABLEornothasattr(context,"abort"):raiseValueError("Invalid gRPC context: missing abort method")status_code:grpc.StatusCode=self._convert_int_to_grpc_status(self.grpc_status)message=self.get_message()ifself.additional_dataandhasattr(context,"set_trailing_metadata"):context.set_trailing_metadata((("additional_data",json.dumps(self.additional_data)),))ifhasattr(context,"abort")andcallable(context.abort):awaitcontext.abort(status_code,message)else:raiseValueError("gRPC context abort method not available or not callable")
defabort_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. """ifcontextisNone:raiseValueError("gRPC context cannot be None")ifnotGRPC_AVAILABLEornothasattr(context,"abort"):raiseValueError("Invalid gRPC context: missing abort method")status_code:grpc.StatusCode=self._convert_int_to_grpc_status(self.grpc_status)message=self.get_message()ifself.additional_dataandhasattr(context,"set_trailing_metadata"):context.set_trailing_metadata((("additional_data",json.dumps(self.additional_data)),))ifhasattr(context,"abort")andcallable(context.abort):context.abort(status_code,message)else:raiseValueError("gRPC context abort method not available or not callable")
@classmethodasyncdefabort_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)awaitinstance.abort_grpc_async(context)
@classmethoddefabort_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)
classUnauthenticatedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse401grpc_status:ClassVar[int]=(StatusCode.UNAUTHENTICATED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAUTHENTICATED.value,tuple)else(StatusCode.UNAUTHENTICATED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse16))
classInvalidCredentialsError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse401grpc_status:ClassVar[int]=(StatusCode.UNAUTHENTICATED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAUTHENTICATED.value,tuple)else(StatusCode.UNAUTHENTICATED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse16))def__init__(self,username:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"username":username}ifusernameelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)defget_message(self)->str:"""Gets the localized error message with username."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enusername=self.additional_data.get("username","username")returntemplate.format(username=username)
defget_message(self)->str:"""Gets the localized error message with username."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enusername=self.additional_data.get("username","username")returntemplate.format(username=username)
classTokenExpiredError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse401grpc_status:ClassVar[int]=(StatusCode.UNAUTHENTICATED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAUTHENTICATED.value,tuple)else(StatusCode.UNAUTHENTICATED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse16))
classInvalidTokenError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse401grpc_status:ClassVar[int]=(StatusCode.UNAUTHENTICATED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAUTHENTICATED.value,tuple)else(StatusCode.UNAUTHENTICATED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse16))
classSessionExpiredError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse401grpc_status:ClassVar[int]=(StatusCode.UNAUTHENTICATED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAUTHENTICATED.value,tuple)else(StatusCode.UNAUTHENTICATED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse16))def__init__(self,session_id:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"session_id":session_id}ifsession_idelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)defget_message(self)->str:"""Gets the localized error message with session ID."""template=self.message_faifself.lang==LanguageType.FAelseself.message_ensession_id=self.additional_data.get("session_id","session_id")returntemplate.format(session_id=session_id)
defget_message(self)->str:"""Gets the localized error message with session ID."""template=self.message_faifself.lang==LanguageType.FAelseself.message_ensession_id=self.additional_data.get("session_id","session_id")returntemplate.format(session_id=session_id)
classPermissionDeniedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneandHTTPStatusisnotNoneelse403)grpc_status:ClassVar[int]=(StatusCode.PERMISSION_DENIED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.PERMISSION_DENIED.value,tuple)else(StatusCode.PERMISSION_DENIED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse7))
classAccountLockedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneandHTTPStatusisnotNoneelse403)grpc_status:ClassVar[int]=(StatusCode.PERMISSION_DENIED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.PERMISSION_DENIED.value,tuple)else(StatusCode.PERMISSION_DENIED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse7))def__init__(self,username:str|None=None,lockout_duration:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifusername:data["username"]=usernameiflockout_duration:data["lockout_duration"]=lockout_durationifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classAccountDisabledError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneandHTTPStatusisnotNoneelse403)grpc_status:ClassVar[int]=(StatusCode.PERMISSION_DENIED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.PERMISSION_DENIED.value,tuple)else(StatusCode.PERMISSION_DENIED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse7))def__init__(self,username:str|None=None,reason:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifusername:data["username"]=usernameifreason:data["reason"]=reasonifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classInvalidVerificationCodeError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneandHTTPStatusisnotNoneelse400)grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneandStatusCodeisnotNoneelse3))def__init__(self,code:str|None=None,remaining_attempts:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifcode:data["code"]=codeifremaining_attemptsisnotNone:data["remaining_attempts"]=remaining_attemptsifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
defget_message(self)->str:"""Gets the localized error message with argument name."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enargument=self.additional_data.get("argument","argument")returntemplate.format(argument=argument)
classInvalidFormatError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,format_type:str|None=None,expected_format:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifformat_type:data["format_type"]=format_typeifexpected_format:data["expected_format"]=expected_formatifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
defget_message(self)->str:"""Gets the localized error message with email."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enemail=self.additional_data.get("email","email")returntemplate.format(email=email)
classInvalidPhoneNumberError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,phone_number:str,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"phone_number":phone_number}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=data)defget_message(self)->str:"""Gets the localized error message with phone number and normalization."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enphone_number=self.additional_data.get("phone_number","phone_number")message=template.format(phone_number=phone_number)# Convert numbers to Persian if language is FAifself.lang==LanguageType.FA:message=StringUtils.convert_english_number_to_persian(message)returnmessage
defget_message(self)->str:"""Gets the localized error message with phone number and normalization."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enphone_number=self.additional_data.get("phone_number","phone_number")message=template.format(phone_number=phone_number)# Convert numbers to Persian if language is FAifself.lang==LanguageType.FA:message=StringUtils.convert_english_number_to_persian(message)returnmessage
classInvalidLandlineNumberError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,landline_number:str,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"landline_number":landline_number}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=data)defget_message(self)->str:"""Gets the localized error message with landline number and normalization."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enlandline_number=self.additional_data.get("landline_number","landline_number")message=template.format(landline_number=landline_number)# Convert numbers to Persian if language is FAifself.lang==LanguageType.FA:message=StringUtils.convert_english_number_to_persian(message)returnmessage
defget_message(self)->str:"""Gets the localized error message with landline number and normalization."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enlandline_number=self.additional_data.get("landline_number","landline_number")message=template.format(landline_number=landline_number)# Convert numbers to Persian if language is FAifself.lang==LanguageType.FA:message=StringUtils.convert_english_number_to_persian(message)returnmessage
classInvalidNationalCodeError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,national_code:str,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"national_code":national_code}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=data)defget_message(self)->str:"""Gets the localized error message with national code and normalization."""template=self.message_faifself.lang==LanguageType.FAelseself.message_ennational_code=self.additional_data.get("national_code","national_code")message=template.format(national_code=national_code)# Convert numbers to Persian if language is FAifself.lang==LanguageType.FA:message=StringUtils.convert_english_number_to_persian(message)returnmessage
defget_message(self)->str:"""Gets the localized error message with national code and normalization."""template=self.message_faifself.lang==LanguageType.FAelseself.message_ennational_code=self.additional_data.get("national_code","national_code")message=template.format(national_code=national_code)# Convert numbers to Persian if language is FAifself.lang==LanguageType.FA:message=StringUtils.convert_english_number_to_persian(message)returnmessage
classInvalidPasswordError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,requirements:list[str]|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"requirements":requirements}ifrequirementselse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classInvalidDateError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,date:str|None=None,expected_format:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifdate:data["date"]=dateifexpected_format:data["expected_format"]=expected_formatifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
defget_message(self)->str:"""Gets the localized error message with URL."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enurl=self.additional_data.get("url","url")returntemplate.format(url=url)
classInvalidIpError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,ip:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"ip":ip}ifipelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)defget_message(self)->str:"""Gets the localized error message with IP address."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enip=self.additional_data.get("ip","ip")returntemplate.format(ip=ip)
defget_message(self)->str:"""Gets the localized error message with IP address."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enip=self.additional_data.get("ip","ip")returntemplate.format(ip=ip)
classInvalidTimestampError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))def__init__(self,timestamp:str|None=None,expected_format:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}iftimestamp:data["timestamp"]=timestampifexpected_format:data["expected_format"]=expected_formatifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classOutOfRangeError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.OUT_OF_RANGE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.OUT_OF_RANGE.value,tuple)else(StatusCode.OUT_OF_RANGE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse11))def__init__(self,field_name:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"field":field_name}iffield_nameelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classNotFoundError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse404grpc_status:ClassVar[int]=(StatusCode.NOT_FOUND.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.NOT_FOUND.value,tuple)else(StatusCode.NOT_FOUND.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse5))def__init__(self,resource_type:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"resource_type":resource_type}ifresource_typeelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)defget_message(self)->str:"""Gets the localized error message with resource type."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enresource_type=self.additional_data.get("resource_type","resource_type")returntemplate.format(resource_type=resource_type)
defget_message(self)->str:"""Gets the localized error message with resource type."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enresource_type=self.additional_data.get("resource_type","resource_type")returntemplate.format(resource_type=resource_type)
classAlreadyExistsError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.ALREADY_EXISTS.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.ALREADY_EXISTS.value,tuple)else(StatusCode.ALREADY_EXISTS.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse6))def__init__(self,resource_type:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"resource_type":resource_type}ifresource_typeelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)defget_message(self)->str:"""Gets the localized error message with resource type."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enresource_type=self.additional_data.get("resource_type","resource_type")returntemplate.format(resource_type=resource_type)
defget_message(self)->str:"""Gets the localized error message with resource type."""template=self.message_faifself.lang==LanguageType.FAelseself.message_enresource_type=self.additional_data.get("resource_type","resource_type")returntemplate.format(resource_type=resource_type)
classConflictError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.ABORTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.ABORTED.value,tuple)else(StatusCode.ABORTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse10))def__init__(self,resource_type:str|None=None,resource_id:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifresource_type:data["resource_type"]=resource_typeifresource_id:data["resource_id"]=resource_idifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classResourceLockedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.ABORTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.ABORTED.value,tuple)else(StatusCode.ABORTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse10))def__init__(self,resource_id:str|None=None,lock_owner:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifresource_id:data["resource_id"]=resource_idiflock_owner:data["lock_owner"]=lock_ownerifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classResourceBusyError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.ABORTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.ABORTED.value,tuple)else(StatusCode.ABORTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse10))def__init__(self,resource_id:str|None=None,busy_reason:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifresource_id:data["resource_id"]=resource_idifbusy_reason:data["busy_reason"]=busy_reasonifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classDataLossError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.DATA_LOSS.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.DATA_LOSS.value,tuple)else(StatusCode.DATA_LOSS.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse15))
classFileTooLargeError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))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={}iffile_name:data["file_name"]=file_nameiffile_size:data["file_size"]=file_sizeifmax_size:data["max_size"]=max_sizeifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classInvalidFileTypeError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse400grpc_status:ClassVar[int]=(StatusCode.INVALID_ARGUMENT.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INVALID_ARGUMENT.value,tuple)else(StatusCode.INVALID_ARGUMENT.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse3))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={}iffile_name:data["file_name"]=file_nameiffile_type:data["file_type"]=file_typeifallowed_types:data["allowed_types"]=allowed_typesifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classQuotaExceededError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse403grpc_status:ClassVar[int]=(StatusCode.RESOURCE_EXHAUSTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.RESOURCE_EXHAUSTED.value,tuple)else(StatusCode.RESOURCE_EXHAUSTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse8))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={}ifquota_type:data["quota_type"]=quota_typeifcurrent_usage:data["current_usage"]=current_usageifquota_limit:data["quota_limit"]=quota_limitifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classResourceExhaustedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse429)grpc_status:ClassVar[int]=(StatusCode.RESOURCE_EXHAUSTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.RESOURCE_EXHAUSTED.value,tuple)else(StatusCode.RESOURCE_EXHAUSTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse8))def__init__(self,resource_type:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"resource_type":resource_type}ifresource_typeelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classStorageError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,storage_type:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"storage_type":storage_type}ifstorage_typeelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classInvalidStateError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.FAILED_PRECONDITION.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.FAILED_PRECONDITION.value,tuple)else(StatusCode.FAILED_PRECONDITION.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse9))def__init__(self,current_state:str|None=None,expected_state:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifcurrent_state:data["current_state"]=current_stateifexpected_state:data["expected_state"]=expected_stateifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classFailedPreconditionError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse412)grpc_status:ClassVar[int]=(StatusCode.FAILED_PRECONDITION.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.FAILED_PRECONDITION.value,tuple)else(StatusCode.FAILED_PRECONDITION.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse9))def__init__(self,precondition:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifprecondition:data["precondition"]=preconditionifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classBusinessRuleViolationError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.FAILED_PRECONDITION.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.FAILED_PRECONDITION.value,tuple)else(StatusCode.FAILED_PRECONDITION.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse9))def__init__(self,rule:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifrule:data["rule"]=ruleifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classInvalidOperationError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse403grpc_status:ClassVar[int]=(StatusCode.PERMISSION_DENIED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.PERMISSION_DENIED.value,tuple)else(StatusCode.PERMISSION_DENIED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse7))def__init__(self,operation:str|None=None,context:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifoperation:data["operation"]=operationifcontext:data["context"]=contextifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classInsufficientFundsError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse402grpc_status:ClassVar[int]=(StatusCode.FAILED_PRECONDITION.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.FAILED_PRECONDITION.value,tuple)else(StatusCode.FAILED_PRECONDITION.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse9))
classInsufficientBalanceError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse402grpc_status:ClassVar[int]=(StatusCode.FAILED_PRECONDITION.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.FAILED_PRECONDITION.value,tuple)else(StatusCode.FAILED_PRECONDITION.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse9))
classMaintenanceModeError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse503)grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))def__init__(self,estimated_duration:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"estimated_duration":estimated_duration}ifestimated_durationelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classNetworkError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse502grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))def__init__(self,service:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifservice:data["service"]=serviceifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classConnectionTimeoutError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse408grpc_status:ClassVar[int]=(StatusCode.DEADLINE_EXCEEDED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.DEADLINE_EXCEEDED.value,tuple)else(StatusCode.DEADLINE_EXCEEDED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse4))def__init__(self,service:str|None=None,timeout:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifservice:data["service"]=serviceiftimeout:data["timeout"]=timeoutifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classServiceUnavailableError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse503)grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))def__init__(self,service:str|None=None,retry_after:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifservice:data["service"]=serviceifretry_after:data["retry_after"]=retry_afterifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classGatewayTimeoutError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse504grpc_status:ClassVar[int]=(StatusCode.DEADLINE_EXCEEDED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.DEADLINE_EXCEEDED.value,tuple)else(StatusCode.DEADLINE_EXCEEDED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse4))def__init__(self,gateway:str|None=None,timeout:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifgateway:data["gateway"]=gatewayiftimeout:data["timeout"]=timeoutifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classBadGatewayError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse502grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))def__init__(self,gateway:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifgateway:data["gateway"]=gatewayifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classRateLimitExceededError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse429)grpc_status:ClassVar[int]=(StatusCode.RESOURCE_EXHAUSTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.RESOURCE_EXHAUSTED.value,tuple)else(StatusCode.RESOURCE_EXHAUSTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse8))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={}ifrate_limit_type:data["rate_limit_type"]=rate_limit_typeifretry_after:data["retry_after"]=retry_afterifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classDatabaseError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,database:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifdatabase:data["database"]=databaseifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classDatabaseConnectionError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse503)grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))
classDatabaseTransactionError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,database:str|None=None,transaction_id:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}iftransaction_id:data["transaction_id"]=transaction_idifadditional_data:data.update(additional_data)super().__init__(database=database,lang=lang,additional_data=dataorNone)
classDatabaseTimeoutError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse408grpc_status:ClassVar[int]=(StatusCode.DEADLINE_EXCEEDED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.DEADLINE_EXCEEDED.value,tuple)else(StatusCode.DEADLINE_EXCEEDED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse4))def__init__(self,database:str|None=None,timeout:int|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}iftimeout:data["timeout"]=timeoutifadditional_data:data.update(additional_data)super().__init__(database=database,lang=lang,additional_data=dataorNone)
classDatabaseDeadlockError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.ABORTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.ABORTED.value,tuple)else(StatusCode.ABORTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse10))
classCacheError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,cache_type:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={}ifcache_type:data["cache_type"]=cache_typeifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classCacheMissError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse404grpc_status:ClassVar[int]=(StatusCode.NOT_FOUND.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.NOT_FOUND.value,tuple)else(StatusCode.NOT_FOUND.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse5))def__init__(self,cache_key:str|None=None,lang:LanguageType|None=None,additional_data:dict|None=None,)->None:data={"cache_key":cache_key}ifcache_keyelse{}ifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)defget_message(self)->str:"""Gets the localized error message with cache key."""template=self.message_faifself.lang==LanguageType.FAelseself.message_encache_key=self.additional_data.get("cache_key","cache_key")returntemplate.format(cache_key=cache_key)
defget_message(self)->str:"""Gets the localized error message with cache key."""template=self.message_faifself.lang==LanguageType.FAelseself.message_encache_key=self.additional_data.get("cache_key","cache_key")returntemplate.format(cache_key=cache_key)
classInternalError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,error_code:str|None=None,lang:LanguageType|None=None,additional_data:dict[str,Any]|None=None,)->None:data={}iferror_code:data["error_code"]=error_codeifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classConfigurationError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,operation:str|None=None,reason:str|None=None,lang:LanguageType|None=None,additional_data:dict[str,Any]|None=None,)->None:data={}ifoperation:data["operation"]=operationifreason:data["reason"]=reasonifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classUnavailableError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse503)grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))def__init__(self,resource_type:str|None=None,lang:LanguageType|None=None,additional_data:dict[str,Any]|None=None,)->None:data={}ifresource_type:data["resource_type"]=resource_typeifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classUnknownError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.UNKNOWN.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNKNOWN.value,tuple)else(StatusCode.UNKNOWN.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse2))def__init__(self,config_key:str|None=None,lang:LanguageType|None=None,additional_data:dict[str,Any]|None=None,)->None:data={}ifconfig_key:data["config_key"]=config_keyifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classAbortedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse409grpc_status:ClassVar[int]=(StatusCode.ABORTED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.ABORTED.value,tuple)else(StatusCode.ABORTED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse10))def__init__(self,service:str|None=None,reason:str|None=None,lang:LanguageType|None=None,additional_data:dict[str,Any]|None=None,)->None:data={}ifservice:data["service"]=serviceifreason:data["reason"]=reasonifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classDeadlockDetectedError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,service:str|None=None,reason:str|None=None,lang:LanguageType|None=None,additional_data:dict[str,Any]|None=None,)->None:data={}ifservice:data["service"]=serviceifreason:data["reason"]=reasonifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classDeadlineExceededError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse408grpc_status:ClassVar[int]=(StatusCode.DEADLINE_EXCEEDED.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.DEADLINE_EXCEEDED.value,tuple)else(StatusCode.DEADLINE_EXCEEDED.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse4))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={}iftimeoutisnotNone:data["timeout"]=timeoutifoperation:data["operation"]=operationifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classDeprecationError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse410grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))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={}ifdeprecated_feature:data["deprecated_feature"]=deprecated_featureifreplacement:data["replacement"]=replacementifremoval_version:data["removal_version"]=removal_versionifadditional_data:data.update(additional_data)super().__init__(lang=lang,additional_data=dataorNone)
classRealmAlreadyExistsError(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]=409grpc_status:ClassVar[int]=6
classUserAlreadyExistsError(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]=409grpc_status:ClassVar[int]=6
classClientAlreadyExistsError(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]=409grpc_status:ClassVar[int]=6
classRoleAlreadyExistsError(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]=409grpc_status:ClassVar[int]=6
classResourceNotFoundError(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]=404grpc_status:ClassVar[int]=5
classInsufficientPermissionsError(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]=403grpc_status:ClassVar[int]=7
classPasswordPolicyError(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]=400grpc_status:ClassVar[int]=3
classKeycloakConnectionTimeoutError(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]=504grpc_status:ClassVar[int]=4
classKeycloakServiceUnavailableError(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]=503grpc_status:ClassVar[int]=14
defget_error_message(keycloak_error:KeycloakError)->str:"""Extract the actual error message from Keycloak error."""error_message=str(keycloak_error)# Try to parse JSON response bodyifhasattr(keycloak_error,"response_body")andkeycloak_error.response_body:try:body=keycloak_error.response_bodybody_str=body.decode("utf-8")ifisinstance(body,bytes)elsestr(body)# body_str is now guaranteed to be str after decodeparsed=json.loads(body_str)ifisinstance(parsed,dict):error_message=(parsed.get("errorMessage")orparsed.get("error_description")orparsed.get("error")orerror_message)exceptjson.JSONDecodeError,UnicodeDecodeError:passreturnerror_message
defhandle_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 datacontext={"original_error":error_message,"response_code":response_code,"keycloak_error_type":type(keycloak_error).__name__,**additional_data,}# Simple string matching to identify error typeserror_lower=error_message.lower()# Realm errorsif"realm"inerror_lowerand"already exists"inerror_lower:returnRealmAlreadyExistsError(additional_data=context)# User errorsif"user exists with same"inerror_lower:returnUserAlreadyExistsError(additional_data=context)# Client errorsif"client"inerror_lowerand"already exists"inerror_lower:returnClientAlreadyExistsError(additional_data=context)# Authentication errorsifany(phraseinerror_lowerforphrasein["invalid user credentials","invalid credentials","authentication failed"]):returnInvalidCredentialsError(additional_data=context)# Not found errorsif"not found"inerror_lower:returnResourceNotFoundError(additional_data=context)# Permission errorsifany(phraseinerror_lowerforphrasein["forbidden","access denied","insufficient permissions"]):returnInsufficientPermissionsError(additional_data=context)# Validation errors (400 status codes that don't match above)ifresponse_code==400:returnValidationError(additional_data=context)# Default to InternalError for unrecognized errorsreturnInternalError(additional_data=context)
classTemporalError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))
classWorkerConnectionError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse503)grpc_status:ClassVar[int]=(StatusCode.UNAVAILABLE.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.UNAVAILABLE.value,tuple)else(StatusCode.UNAVAILABLE.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse14))def__init__(self,additional_data:dict[str,Any]|None=None,)->None:"""Initialize the worker connection error."""super().__init__(additional_data=additional_data)
classWorkerShutdownError(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.valueifHTTP_AVAILABLEandHTTPStatusisnotNoneelse500)grpc_status:ClassVar[int]=(StatusCode.INTERNAL.value[0]ifGRPC_AVAILABLEandStatusCodeisnotNoneandisinstance(StatusCode.INTERNAL.value,tuple)else(StatusCode.INTERNAL.valueifGRPC_AVAILABLEandStatusCodeisnotNoneelse13))def__init__(self,additional_data:dict[str,Any]|None=None,)->None:"""Initialize the worker shutdown error."""super().__init__(additional_data=additional_data)