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
|
import os
from typing import Optional
from shared.abstractions import R2RException
class BaseClient:
def __init__(
self,
base_url: str | None = None,
timeout: float = 300.0,
):
self.base_url = base_url or os.getenv(
"R2R_API_BASE", "https://api.sciphi.ai"
)
self.timeout = timeout
self.access_token: Optional[str] = None
self._refresh_token: Optional[str] = None
self._user_id: Optional[str] = None
self.api_key: Optional[str] = os.getenv("R2R_API_KEY", None)
def _get_auth_header(self) -> dict[str, str]:
if self.access_token and self.api_key:
raise R2RException(
status_code=400,
message="Cannot have both access token and api key.",
)
if self.access_token:
return {"Authorization": f"Bearer {self.access_token}"}
elif self.api_key:
return {"x-api-key": self.api_key}
else:
return {}
def _get_full_url(self, endpoint: str, version: str = "v3") -> str:
return f"{self.base_url}/{version}/{endpoint}"
def _prepare_request_args(self, endpoint: str, **kwargs) -> dict:
headers = kwargs.pop("headers", {})
if (self.access_token or self.api_key) and endpoint not in [
"register",
"login",
"verify_email",
]:
headers.update(self._get_auth_header())
if (
kwargs.get("params", None) == {}
or kwargs.get("params", None) is None
):
kwargs.pop("params", None)
return {"headers": headers, **kwargs}
|
