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
|
from __future__ import annotations
from typing import Any, Callable, Dict, Optional, TypeVar, overload
from httpx import Response
from pydantic import BaseModel
from typing_extensions import Literal, Self
from ..constants import API_VERSION_HEADER_NAME, API_VERSIONS
from ..helpers import handle_exception, model_dump
from ..http_clients import SyncClient
T = TypeVar("T")
class SyncGoTrueBaseAPI:
def __init__(
self,
*,
url: str,
headers: Dict[str, str],
http_client: Optional[SyncClient],
verify: bool = True,
proxy: Optional[str] = None,
):
self._url = url
self._headers = headers
self._http_client = http_client or SyncClient(
verify=bool(verify),
proxy=proxy,
follow_redirects=True,
http2=True,
)
def __enter__(self) -> Self:
return self
def __exit__(self, exc_t, exc_v, exc_tb) -> None:
self.close()
def close(self) -> None:
self._http_client.aclose()
@overload
def _request(
self,
method: Literal["GET", "OPTIONS", "HEAD", "POST", "PUT", "PATCH", "DELETE"],
path: str,
*,
jwt: Optional[str] = None,
redirect_to: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
query: Optional[Dict[str, str]] = None,
body: Optional[Any] = None,
no_resolve_json: Literal[False] = False,
xform: Callable[[Any], T],
) -> T: ... # pragma: no cover
@overload
def _request(
self,
method: Literal["GET", "OPTIONS", "HEAD", "POST", "PUT", "PATCH", "DELETE"],
path: str,
*,
jwt: Optional[str] = None,
redirect_to: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
query: Optional[Dict[str, str]] = None,
body: Optional[Any] = None,
no_resolve_json: Literal[True],
xform: Callable[[Response], T],
) -> T: ... # pragma: no cover
@overload
def _request(
self,
method: Literal["GET", "OPTIONS", "HEAD", "POST", "PUT", "PATCH", "DELETE"],
path: str,
*,
jwt: Optional[str] = None,
redirect_to: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
query: Optional[Dict[str, str]] = None,
body: Optional[Any] = None,
no_resolve_json: bool = False,
) -> None: ... # pragma: no cover
def _request(
self,
method: Literal["GET", "OPTIONS", "HEAD", "POST", "PUT", "PATCH", "DELETE"],
path: str,
*,
jwt: Optional[str] = None,
redirect_to: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
query: Optional[Dict[str, str]] = None,
body: Optional[Any] = None,
no_resolve_json: bool = False,
xform: Optional[Callable[[Any], T]] = None,
) -> Optional[T]:
url = f"{self._url}/{path}"
headers = {**self._headers, **(headers or {})}
if API_VERSION_HEADER_NAME not in headers:
headers[API_VERSION_HEADER_NAME] = API_VERSIONS["2024-01-01"].get("name")
if "Content-Type" not in headers:
headers["Content-Type"] = "application/json;charset=UTF-8"
if jwt:
headers["Authorization"] = f"Bearer {jwt}"
query = query or {}
if redirect_to:
query["redirect_to"] = redirect_to
try:
response = self._http_client.request(
method,
url,
headers=headers,
params=query,
json=model_dump(body) if isinstance(body, BaseModel) else body,
)
response.raise_for_status()
result = response if no_resolve_json else response.json()
if xform:
return xform(result)
except Exception as e:
raise handle_exception(e)
|
