Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2024-12-23 03:33:35 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Add optional error code to exceptions (#585)

Browse Source
This commit is contained in:
sdb9696 2023-12-20 19:16:23 +00:00 committed by GitHub
parent 6819c746d7
commit b66347116f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 22 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
kasa/aestransport.py
View File

@ -130,14 +130,14 @@ class AesTransport(BaseTransport):
return return
msg = f"{msg}: {self._host}: {error_code.name}({error_code.value})" msg = f"{msg}: {self._host}: {error_code.name}({error_code.value})"
if error_code in SMART_TIMEOUT_ERRORS: if error_code in SMART_TIMEOUT_ERRORS:
raise TimeoutException(msg) raise TimeoutException(msg, error_code=error_code)
if error_code in SMART_RETRYABLE_ERRORS: if error_code in SMART_RETRYABLE_ERRORS:
raise RetryableException(msg) raise RetryableException(msg, error_code=error_code)
if error_code in SMART_AUTHENTICATION_ERRORS: if error_code in SMART_AUTHENTICATION_ERRORS:
self._handshake_done = False self._handshake_done = False
self._login_token = None self._login_token = None
raise AuthenticationException(msg) raise AuthenticationException(msg, error_code=error_code)
raise SmartDeviceException(msg) raise SmartDeviceException(msg, error_code=error_code)
async def send_secure_passthrough(self, request: str): async def send_secure_passthrough(self, request: str):
"""Send encrypted message as passthrough.""" """Send encrypted message as passthrough."""

14
kasa/exceptions.py
View File

@ -1,10 +1,15 @@
"""python-kasa exceptions.""" """python-kasa exceptions."""
from enum import IntEnum from enum import IntEnum
from typing import Optional
class SmartDeviceException(Exception): class SmartDeviceException(Exception):
"""Base exception for device errors.""" """Base exception for device errors."""
def __init__(self, *args, error_code: Optional["SmartErrorCode"] = None):
self.error_code = error_code
super().__init__(args)
class UnsupportedDeviceException(SmartDeviceException): class UnsupportedDeviceException(SmartDeviceException):
"""Exception for trying to connect to unsupported devices.""" """Exception for trying to connect to unsupported devices."""
@ -17,14 +22,23 @@ class UnsupportedDeviceException(SmartDeviceException):
class AuthenticationException(SmartDeviceException): class AuthenticationException(SmartDeviceException):
"""Base exception for device authentication errors.""" """Base exception for device authentication errors."""
def __init__(self, *args, error_code: Optional["SmartErrorCode"] = None):
super().__init__(args, error_code)
class RetryableException(SmartDeviceException): class RetryableException(SmartDeviceException):
"""Retryable exception for device errors.""" """Retryable exception for device errors."""
def __init__(self, *args, error_code: Optional["SmartErrorCode"] = None):
super().__init__(args, error_code)
class TimeoutException(SmartDeviceException): class TimeoutException(SmartDeviceException):
"""Timeout exception for device errors.""" """Timeout exception for device errors."""
def __init__(self, *args, error_code: Optional["SmartErrorCode"] = None):
super().__init__(args, error_code)
class SmartErrorCode(IntEnum): class SmartErrorCode(IntEnum):
"""Enum for SMART Error Codes.""" """Enum for SMART Error Codes."""

8
kasa/smartprotocol.py
View File

@ -186,12 +186,12 @@ class SmartProtocol(TPLinkProtocol):
if method := resp_dict.get("method"): if method := resp_dict.get("method"):
msg += f" for method: {method}" msg += f" for method: {method}"
if error_code in SMART_TIMEOUT_ERRORS: if error_code in SMART_TIMEOUT_ERRORS:
raise TimeoutException(msg) raise TimeoutException(msg, error_code=error_code)
if error_code in SMART_RETRYABLE_ERRORS: if error_code in SMART_RETRYABLE_ERRORS:
raise RetryableException(msg) raise RetryableException(msg, error_code=error_code)
if error_code in SMART_AUTHENTICATION_ERRORS: if error_code in SMART_AUTHENTICATION_ERRORS:
raise AuthenticationException(msg) raise AuthenticationException(msg, error_code=error_code)
raise SmartDeviceException(msg) raise SmartDeviceException(msg, error_code=error_code)
async def close(self) -> None: async def close(self) -> None:
"""Close the protocol.""" """Close the protocol."""