|
3 | 3 | import secrets |
4 | 4 | import time |
5 | 5 | from importlib import resources |
| 6 | +from importlib.metadata import version |
6 | 7 | from pathlib import Path |
7 | 8 |
|
8 | 9 | import httpx |
9 | 10 | from authlib.jose import jwt |
10 | 11 | from authlib.oidc.core import IDToken |
11 | 12 | from httpx import URL |
12 | | -from pydantic import parse_obj_as |
| 13 | +from pydantic import TypeAdapter |
13 | 14 |
|
14 | 15 | from . import errors, keys |
15 | 16 | from . import models as m |
16 | 17 | from .queries import load_query |
17 | 18 |
|
18 | 19 | CLIENT_ID = "ONEKEY Python SDK" |
19 | 20 | TOKEN_NAMESPACE = "https://www.onekey.com/" # noqa: S105 (hardcoded credential) |
| 21 | +APP_NAME = "onekey_client" |
| 22 | +APP_VERSION = version(APP_NAME) |
20 | 23 |
|
21 | 24 |
|
22 | 25 | def _login_required(func): |
@@ -63,17 +66,18 @@ def _setup_httpx_client( |
63 | 66 | ca_bundle: Path | None = None, |
64 | 67 | disable_tls_verify: bool | None = False, |
65 | 68 | ): |
| 69 | + headers = {"User-Agent": f"{APP_NAME}/{APP_VERSION}"} |
66 | 70 | if disable_tls_verify: |
67 | | - return httpx.Client(base_url=api_url, verify=False) # noqa: S501 (TLS certificate validation disabled) |
| 71 | + return httpx.Client(base_url=api_url, headers=headers, verify=False) # noqa: S501 (TLS certificate validation disabled) |
68 | 72 |
|
69 | 73 | if ca_bundle is not None: |
70 | 74 | ca = ca_bundle.expanduser() |
71 | 75 | if not ca.exists(): |
72 | 76 | raise errors.InvalidCABundle |
73 | 77 |
|
74 | | - return httpx.Client(base_url=api_url, verify=str(ca)) |
| 78 | + return httpx.Client(base_url=api_url, headers=headers, verify=str(ca)) |
75 | 79 | with resources.path(keys, "ca.pem") as ca: |
76 | | - return httpx.Client(base_url=api_url, verify=str(ca)) |
| 80 | + return httpx.Client(base_url=api_url, headers=headers, verify=str(ca)) |
77 | 81 |
|
78 | 82 | def _load_key(self, key_name: str, path: Path | None = None): |
79 | 83 | if path is not None: |
@@ -103,7 +107,7 @@ def login(self, email: str, password: str): |
103 | 107 | claims_cls=IDToken, |
104 | 108 | ) |
105 | 109 | tenants = id_token[TOKEN_NAMESPACE + "tenants"] |
106 | | - tenants = parse_obj_as(list[m.Tenant], tenants) |
| 110 | + tenants = TypeAdapter(list[m.Tenant]).validate_python(tenants) |
107 | 111 | self._state.tenants = {e.name: e for e in tenants} |
108 | 112 | self._state.email = email |
109 | 113 | self._state.raw_id_token = json_res["id_token"] |
|
0 commit comments