-
Notifications
You must be signed in to change notification settings - Fork 79
/
client.py
99 lines (74 loc) · 3.94 KB
/
client.py
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
"""Generic API management."""
import logging
from collections import defaultdict, deque
from dataclasses import dataclass
from typing import Deque, Dict, Optional
import httpx
from bimmer_connected.api.authentication import MyBMWAuthentication
from bimmer_connected.api.regions import get_app_version, get_server_url, get_user_agent
from bimmer_connected.api.utils import anonymize_response, get_correlation_id, handle_httpstatuserror
from bimmer_connected.const import HTTPX_TIMEOUT, X_USER_AGENT, CarBrands
from bimmer_connected.models import AnonymizedResponse, GPSPosition
_LOGGER = logging.getLogger(__name__)
RESPONSE_STORE: Deque[AnonymizedResponse] = deque(maxlen=10)
@dataclass
class MyBMWClientConfiguration:
"""Stores global settings for MyBMWClient."""
authentication: MyBMWAuthentication
log_responses: Optional[bool] = False
observer_position: Optional[GPSPosition] = None
verify: httpx._types.VerifyTypes = True
def set_log_responses(self, log_responses: bool) -> None:
"""Set if responses are logged and clear response store."""
self.log_responses = log_responses
RESPONSE_STORE.clear()
class MyBMWClient(httpx.AsyncClient):
"""Async HTTP client based on `httpx.AsyncClient` with automated OAuth token refresh."""
def __init__(self, config: MyBMWClientConfiguration, *args, brand: Optional[CarBrands] = None, **kwargs):
self.config = config
# Add authentication
kwargs["auth"] = self.config.authentication
# Increase timeout
kwargs["timeout"] = httpx.Timeout(HTTPX_TIMEOUT)
# Use external SSL context stored in MyBMWClientConfiguration. Required in Home Assistant due to event loop
# blocking when httpx loads SSL certificates from disk. If not given, uses httpx defaults.
kwargs["verify"] = self.config.verify
# Set default values
kwargs["base_url"] = kwargs.get("base_url") or get_server_url(config.authentication.region)
kwargs["headers"] = kwargs.get("headers") or self.generate_default_header(brand)
# Register event hooks
kwargs["event_hooks"] = defaultdict(list, **kwargs.get("event_hooks", {}))
# Event hook for logging content
async def log_response(response: httpx.Response):
await response.aread()
RESPONSE_STORE.append(anonymize_response(response))
if config.log_responses:
kwargs["event_hooks"]["response"].append(log_response)
# Event hook which calls raise_for_status on all requests
async def raise_for_status_event_handler(response: httpx.Response):
"""Event handler that automatically raises HTTPStatusErrors when attached.
Will only raise on 4xx/5xx errors but not 401/429 which are handled `self.auth`.
"""
if response.is_error and response.status_code not in [401, 429]:
try:
response.raise_for_status()
except httpx.HTTPStatusError as ex:
await handle_httpstatuserror(ex, log_handler=_LOGGER)
kwargs["event_hooks"]["response"].append(raise_for_status_event_handler)
super().__init__(*args, **kwargs)
def generate_default_header(self, brand: Optional[CarBrands] = None) -> Dict[str, str]:
"""Generate a header for HTTP requests to the server."""
return {
"accept": "application/json",
"accept-language": "en",
"x-raw-locale": "en-US",
"user-agent": get_user_agent(self.config.authentication.region),
"x-user-agent": X_USER_AGENT.format(
brand=(brand or CarBrands.BMW).value,
app_version=get_app_version(self.config.authentication.region),
region=self.config.authentication.region.value,
),
**get_correlation_id(),
"bmw-units-preferences": "d=KM;v=L;p=B;ec=KWH100KM;fc=L100KM;em=GKM;",
"24-hour-format": "true",
}