File: error_handling.py

package info (click to toggle)
pycontrol4 2.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 656 kB
  • sloc: python: 1,438; makefile: 3
file content (164 lines) | stat: -rw-r--r-- 4,661 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"""Handles errors received from the Control4 API."""

from __future__ import annotations

from typing import Any

import json
import xmltodict


class C4Exception(Exception):
    """Base error for pyControl4."""

    def __init__(self, message: str) -> None:
        self.message = message


class NotFound(C4Exception):
    """Raised when a 404 response is received from the Control4 API.
    Occurs when the requested controller, etc. could not be found."""


class Unauthorized(C4Exception):
    """Raised when unauthorized, but no other recognized details are provided.
    Occurs when token is invalid."""


class BadCredentials(Unauthorized):
    """Raised when provided credentials are incorrect."""


class BadToken(Unauthorized):
    """Raised when director bearer token is invalid."""


class InvalidCategory(C4Exception):
    """Raised when an invalid category is provided when calling
    `pyControl4.director.C4Director.get_all_items_by_category`."""


ERROR_CODES = {"401": Unauthorized, "404": NotFound}

ERROR_DETAILS = {
    "Permission denied Bad credentials": BadCredentials,
}

DIRECTOR_ERRORS = {"Unauthorized": Unauthorized, "Invalid category": InvalidCategory}

DIRECTOR_ERROR_DETAILS = {"Expired or invalid token": BadToken}


def _check_response_format(response_text: str) -> str:
    """Known Control4 authentication API error message formats:
    ```json
    {
        "C4ErrorResponse": {
            "code": 401,
            "details": "Permission denied Bad credentials",
            "message": "Permission denied",
            "subCode": 0
        }
    }
    ```
    ```json
    {
        "code": 404,
        "details": "Account with id:000000 not found in DB",
        "message": "Account not found",
        "subCode": 0
    }```
    ```xml
    <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
    <C4ErrorResponse>
        <code>401</code>
        <details></details>
        <message>Permission denied</message>
        <subCode>0</subCode>
    </C4ErrorResponse>
    ```
    Known Control4 director error message formats:
    ```json
    {
        "error": "Unauthorized",
        "details": "Expired or invalid token"
    }
    ```
    """
    if response_text.startswith("<"):
        return "XML"
    return "JSON"


def _extract_error_info(dictionary: dict[str, Any]) -> dict[str, Any] | None:
    """Extract error information from a parsed Control4 response.

    Returns a dict with 'details', 'code', or 'error' key, or None if no error found.
    """
    # Check for C4ErrorResponse format
    if "C4ErrorResponse" in dictionary:
        error_resp = dictionary.get("C4ErrorResponse", {})
        return {
            "details": error_resp.get("details"),
            "code": error_resp.get("code"),
            "type": "C4ErrorResponse",
        }

    # Check for direct code format
    if "code" in dictionary:
        return {
            "details": dictionary.get("details"),
            "code": dictionary.get("code"),
            "type": "code",
        }

    # Check for error format (director)
    if "error" in dictionary:
        return {
            "details": dictionary.get("details"),
            "error": dictionary.get("error"),
            "type": "error",
        }

    return None


def _raise_error(error_info: dict[str, Any], response_text: str) -> None:
    """Raise appropriate exception based on error info."""
    details = error_info.get("details")
    code = error_info.get("code")
    error = error_info.get("error")

    # Try to match by details first (most specific)
    if details:
        if details in ERROR_DETAILS:
            raise ERROR_DETAILS[details](response_text)
        if details in DIRECTOR_ERROR_DETAILS:
            raise DIRECTOR_ERROR_DETAILS[details](response_text)

    # Try to match by code/error (less specific)
    if code is not None:
        raise ERROR_CODES.get(str(code), C4Exception)(response_text)
    if error is not None:
        raise DIRECTOR_ERRORS.get(str(error), C4Exception)(response_text)

    # If nothing matched, raise generic error
    raise C4Exception(response_text)


def check_response_for_error(response_text: str) -> None:
    """Checks a string response from the Control4 API for error codes.

    Parameters:
        `response_text` - JSON or XML response from Control4, as a string.
    """
    response_format = _check_response_format(response_text)

    if response_format == "JSON":
        dictionary = json.loads(response_text)
    else:  # XML
        dictionary = xmltodict.parse(response_text)

    error_info = _extract_error_info(dictionary)
    if error_info:
        _raise_error(error_info, response_text)