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
|
"""API management class and base class for the different end points."""
from abc import ABC
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, TypedDict, TypeVar
import orjson
from ..errors import (
AiounifiException,
LoginRequired,
NoPermission,
TwoFaTokenRequired,
Unauthorized,
)
ERRORS = {
"api.err.Invalid": Unauthorized,
"api.err.LoginRequired": LoginRequired,
"api.err.NoPermission": NoPermission,
"api.err.Ubic2faTokenRequired": TwoFaTokenRequired,
}
class TypedApiResponse(TypedDict, total=False):
"""Common response."""
meta: dict[str, Any]
data: list[dict[str, Any]]
@dataclass
class ApiRequest:
"""Data class with required properties of a request."""
method: str
path: str
data: Mapping[str, Any] | None = None
def full_path(self, site: str, is_unifi_os: bool) -> str:
"""Create url to work with a specific controller."""
if is_unifi_os:
return f"/proxy/network/api/s/{site}{self.path}"
return f"/api/s/{site}{self.path}"
def decode(self, raw: bytes) -> TypedApiResponse:
"""Put data, received from the unifi controller, into a TypedApiResponse."""
data: TypedApiResponse = orjson.loads(raw)
if "meta" in data and data["meta"]["rc"] == "error":
raise ERRORS.get(data["meta"]["msg"], AiounifiException)(data)
return data
@dataclass
class ApiRequestV2(ApiRequest):
"""Data class with required properties of a V2 API request."""
def full_path(self, site: str, is_unifi_os: bool) -> str:
"""Create url to work with a specific controller."""
if is_unifi_os:
return f"/proxy/network/v2/api/site/{site}{self.path}"
return f"/v2/api/site/{site}{self.path}"
def decode(self, raw: bytes) -> TypedApiResponse:
"""Put data, received from the unifi controller, into a TypedApiResponse."""
data = orjson.loads(raw)
if "errorCode" in data:
raise ERRORS.get(data["message"], AiounifiException)(data)
return TypedApiResponse(
meta={"rc": "ok", "msg": ""},
data=data if isinstance(data, list) else [data],
)
# @dataclass
# class ApiItem(ABC):
# """Base class for all end points using APIItems class."""
# raw: Any
class ApiItem(ABC):
"""Base class for all end points using APIItems class."""
def __init__(self, raw: Any) -> None:
"""Initialize API item."""
self.raw = raw
ApiItemT = TypeVar("ApiItemT", bound=ApiItem)
|
