Repository navigation
Expand file tree
/
Copy pathclient.py
More file actions
129 lines (103 loc) · 4.4 KB
/
Copy pathclient.py
File metadata and controls
129 lines (103 loc) · 4.4 KB
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
126
127
128
129
"""Top-level :class:`StemSplit` client and the placeholder
:class:`AsyncStemSplit` that points at the v0.2 milestone.
"""
from __future__ import annotations
import os
import warnings
from typing import TYPE_CHECKING
from stemsplit_python._transport import (
DEFAULT_BASE_URL,
DEFAULT_MAX_RETRIES,
RateLimit,
Transport,
)
from stemsplit_python.resources.account import AccountResource
from stemsplit_python.resources.jobs import JobsResource
from stemsplit_python.resources.uploads import UploadsResource
from stemsplit_python.resources.webhooks import WebhooksResource
from stemsplit_python.resources.youtube_jobs import YouTubeJobsResource
if TYPE_CHECKING:
import httpx
class StemSplit:
"""Synchronous client for the StemSplit API.
Args:
api_key: The API key. Defaults to the ``STEMSPLIT_API_KEY`` environment
variable. Strings beginning with anything other than ``sk_live_``
or ``sk_test_`` produce a warning, not an error.
base_url: API base URL. Defaults to ``https://stemsplit.io/api/v1``.
timeout: Per-request timeout. Defaults to a sane mix (10s connect,
30s read for normal calls; 10min for uploads).
max_retries: Number of retries on idempotent verbs and ``429``/``5xx``.
Honors ``Retry-After``. Defaults to ``3``.
http_client: Optional pre-built :class:`httpx.Client`. The client
takes ownership of the underlying client only when it constructs
it itself; user-supplied clients are not closed.
Example::
from pathlib import Path
from stemsplit_python import StemSplit
client = StemSplit(api_key="sk_live_...")
job = client.jobs.create(audio=Path("song.mp3"), output_type="BOTH")
job.wait()
job.download_all("./out/")
"""
def __init__(
self,
*,
api_key: str | None = None,
base_url: str | None = None,
timeout: float | None = None,
max_retries: int = DEFAULT_MAX_RETRIES,
http_client: httpx.Client | None = None,
) -> None:
resolved_key = api_key if api_key is not None else os.environ.get("STEMSPLIT_API_KEY")
if not resolved_key:
raise ValueError(
"Missing API key. Pass api_key= or set the STEMSPLIT_API_KEY "
"environment variable. Get one at https://stemsplit.io/app/settings/api"
)
if not (resolved_key.startswith("sk_live_") or resolved_key.startswith("sk_test_")):
warnings.warn(
"API key does not begin with 'sk_live_' or 'sk_test_'. The server "
"may reject requests — check your key in the dashboard.",
stacklevel=2,
)
self._transport = Transport(
api_key=resolved_key,
base_url=base_url or DEFAULT_BASE_URL,
timeout=timeout,
max_retries=max_retries,
http_client=http_client,
)
self.uploads = UploadsResource(self._transport)
self.jobs = JobsResource(self._transport, self.uploads)
self.youtube_jobs = YouTubeJobsResource(self._transport)
self.account = AccountResource(self._transport)
self.webhooks = WebhooksResource(self._transport)
@property
def base_url(self) -> str:
return self._transport.base_url
@property
def last_rate_limit(self) -> RateLimit | None:
"""Latest ``X-RateLimit-*`` headers parsed from the most recent response."""
return self._transport.last_rate_limit
def close(self) -> None:
"""Close the underlying HTTP client (only if the SDK created it)."""
self._transport.close()
def __enter__(self) -> StemSplit:
return self
def __exit__(self, *exc: object) -> None:
self.close()
class AsyncStemSplit:
"""Placeholder for the async client.
The async surface ships in v0.2 — track progress on
`https://github.com/StemSplit/stemsplit-python/issues`. Constructing
this class today raises :class:`NotImplementedError`; use the sync
:class:`StemSplit` in the meantime.
"""
def __init__(self, *_: object, **__: object) -> None:
raise NotImplementedError(
"AsyncStemSplit lands in v0.2. Track the milestone at "
"https://github.com/StemSplit/stemsplit-python/issues — for now use "
"the sync StemSplit client."
)
__all__ = ["AsyncStemSplit", "StemSplit"]