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
|
"""Helper functions for pytedee_async."""
import asyncio
from http import HTTPStatus
from typing import Any, Mapping
import aiohttp
from .const import API_URL_DEVICE, TIMEOUT
from .exception import TedeeAuthException, TedeeClientException, TedeeRateLimitException
async def is_personal_key_valid(
personal_key: str,
session: aiohttp.ClientSession,
timeout: int = TIMEOUT,
) -> bool:
"""Check if personal key is valid."""
try:
response = await session.get(
API_URL_DEVICE,
headers={
"Content-Type": "application/json",
"Authorization": "PersonalKey " + personal_key,
},
timeout=timeout,
)
except (aiohttp.ClientError, aiohttp.ServerConnectionError, TimeoutError):
return False
await asyncio.sleep(0.1)
if response.status in (
HTTPStatus.OK,
HTTPStatus.CREATED,
HTTPStatus.ACCEPTED,
):
return True
return False
async def http_request(
url: str,
http_method: str,
headers: Mapping[str, str] | None,
session: aiohttp.ClientSession,
timeout: int = TIMEOUT,
json_data: Any = None,
) -> Any:
"""HTTP request wrapper."""
try:
response = await session.request(
http_method,
url,
headers=headers,
json=json_data,
timeout=timeout,
)
except (
aiohttp.ServerConnectionError,
aiohttp.ClientError,
TimeoutError,
) as exc:
raise TedeeClientException(f"Error during http call: {exc}") from exc
await asyncio.sleep(0.1)
status_code = response.status
if response.status in (
HTTPStatus.OK,
HTTPStatus.CREATED,
HTTPStatus.ACCEPTED,
HTTPStatus.NO_CONTENT,
):
return await response.json()
if status_code == HTTPStatus.UNAUTHORIZED:
raise TedeeAuthException("Authentication failed.")
if status_code == HTTPStatus.TOO_MANY_REQUESTS:
raise TedeeRateLimitException("Tedee API Rate Limit.")
if status_code == HTTPStatus.NOT_FOUND:
raise TedeeClientException("Resource not found.")
if status_code == HTTPStatus.NOT_ACCEPTABLE:
raise TedeeClientException("Request not acceptable.")
if status_code == HTTPStatus.CONFLICT:
raise TedeeClientException("Conflict.")
raise TedeeClientException(f"Error during HTTP request. Status code {status_code}")
|
