Skip to content

Add: Added timeout for request calls in Music AI #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
humbertonc merged 3 commits into from
Jul 17, 2026
+133 −32
Merged

Add: Added timeout for request calls in Music AI #8

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d7b8619
Add: Added timeout for request calls in Music AI
Jul 16, 2026
243238c
Add: Integration tests for new timeout functionality
Jul 17, 2026
135e7e9
Add: Bump version
Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 72 additions & 30 deletions musicai_sdk/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,15 @@
get_user_agent,
)

# Sentinel used to distinguish "no per-call timeout given" (fall back to the
# instance default) from an explicit `timeout=None` (disable the timeout).
class _Unset:
def __repr__(self):
return "UNSET"


_UNSET = _Unset()


class MusicAiClient:
def __init__(
Expand All @@ -20,55 +29,74 @@ def __init__(
job_monitor_interval=2,
save_output_to_folder=True,
user_agent=None,
timeout=None,
):
self.api_key = api_key
self.base_url = "https://api.music.ai/v1"
self.job_monitor_interval = job_monitor_interval
self.save_output_to_folder = save_output_to_folder
self.user_agent = user_agent or get_user_agent()
self.timeout = timeout

def get_headers(self):
return {
"Authorization": self.api_key,
"User-Agent": self.user_agent,
}

def upload_file(self, file_path):
response = requests.get(f"{self.base_url}/upload", headers=self.get_headers())
def _request(self, method, url, timeout=_UNSET, **kwargs):
resolved_timeout = self.timeout if timeout is _UNSET else timeout
return requests.request(method, url, timeout=resolved_timeout, **kwargs)

def upload_file(self, file_path, timeout=_UNSET):
response = self._request(
"get",
f"{self.base_url}/upload",
headers=self.get_headers(),
timeout=timeout,
)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error uploading file: {response.status_code} {response.text}"
)
upload_url = response.json()["uploadUrl"]
with open(file_path, "rb") as file:
put_response = requests.put(upload_url, data=file.read())
put_response = self._request(
"put", upload_url, data=file.read(), timeout=timeout
)
if put_response.status_code // 100 != 2:
raise HTTPError(
f"Error uploading file to provided URL: {put_response.status_code} {put_response.text}"
)
return response.json()["downloadUrl"]

def get_job(self, job_id):
response = requests.get(
f"{self.base_url}/job/{job_id}", headers=self.get_headers()
def get_job(self, job_id, timeout=_UNSET):
response = self._request(
"get",
f"{self.base_url}/job/{job_id}",
headers=self.get_headers(),
timeout=timeout,
)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error getting job: {response.status_code} {response.text}"
)
return response.json()

def get_job_status(self, job_id):
response = requests.get(
f"{self.base_url}/job/{job_id}/status", headers=self.get_headers()
def get_job_status(self, job_id, timeout=_UNSET):
response = self._request(
"get",
f"{self.base_url}/job/{job_id}/status",
headers=self.get_headers(),
timeout=timeout,
)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error getting job: {response.status_code} {response.text}"
)
return response.json()

def list_jobs(self, filters=None):
def list_jobs(self, filters=None, timeout=_UNSET):
params = {}

if filters:
Expand All @@ -84,7 +112,9 @@ def list_jobs(self, filters=None):
if query_string
else f"{self.base_url}/job"
)
response = requests.get(url, headers=self.get_headers())
response = self._request(
"get", url, headers=self.get_headers(), timeout=timeout
)

if response.status_code // 100 != 2:
raise HTTPError(
Expand All @@ -93,7 +123,7 @@ def list_jobs(self, filters=None):

return response.json()

def add_job(self, job_name, workflow_slug, params, **options):
def add_job(self, job_name, workflow_slug, params, timeout=_UNSET, **options):
data = {
"name": job_name,
"workflow": workflow_slug,
Expand All @@ -106,26 +136,33 @@ def add_job(self, job_name, workflow_slug, params, **options):
if "copy_results_to" in options:
data["copyResultsTo"] = options["copy_results_to"]

response = requests.post(
f"{self.base_url}/job", headers=self.get_headers(), json=data
response = self._request(
"post",
f"{self.base_url}/job",
headers=self.get_headers(),
json=data,
timeout=timeout,
)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error creating job: {response.status_code} {response.text}"
)
return response.json()

def delete_job(self, job_id):
response = requests.delete(
f"{self.base_url}/job/{job_id}", headers=self.get_headers()
def delete_job(self, job_id, timeout=_UNSET):
response = self._request(
"delete",
f"{self.base_url}/job/{job_id}",
headers=self.get_headers(),
timeout=timeout,
)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error deleting job: {response.status_code} {response.text}"
)
return response.json()

def list_workflows(self, filters=None):
def list_workflows(self, filters=None, timeout=_UNSET):
params = {}

if filters:
Expand All @@ -141,7 +178,9 @@ def list_workflows(self, filters=None):
if query_string
else f"{self.base_url}/workflow"
)
response = requests.get(url, headers=self.get_headers())
response = self._request(
"get", url, headers=self.get_headers(), timeout=timeout
)

if response.status_code // 100 != 2:
raise HTTPError(
Expand All @@ -150,26 +189,29 @@ def list_workflows(self, filters=None):

return response.json()

def wait_for_job_completion(self, id):
def wait_for_job_completion(self, id, timeout=_UNSET):
while True:
job = self.get_job_status(id)
job = self.get_job_status(id, timeout=timeout)
if job["status"] in ["SUCCEEDED", "FAILED"]:
return self.get_job(id)
return self.get_job(id, timeout=timeout)
time.sleep(self.job_monitor_interval),

def get_application_info(self):
response = requests.get(
f"{self.base_url}/application", headers=self.get_headers()
def get_application_info(self, timeout=_UNSET):
response = self._request(
"get",
f"{self.base_url}/application",
headers=self.get_headers(),
timeout=timeout,
)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error getting application info: {response.status_code} {response.text}"
)
return response.json()

def download_job_results(self, job_id_or_job_data, output_dir):
def download_job_results(self, job_id_or_job_data, output_dir, timeout=_UNSET):
job = (
self.get_job(job_id_or_job_data)
self.get_job(job_id_or_job_data, timeout=timeout)
if isinstance(job_id_or_job_data, str)
else job_id_or_job_data
)
Expand All @@ -192,7 +234,7 @@ def download_job_results(self, job_id_or_job_data, output_dir):
else extract_name_from_url(value)
)
download_destination = os.path.join(output_dir, file_name)
self.download_file(value, download_destination)
self.download_file(value, download_destination, timeout=timeout)
download_result[result] = download_destination
result_json_downloads[result] = f"./{file_name}"

Expand All @@ -202,9 +244,9 @@ def download_job_results(self, job_id_or_job_data, output_dir):

return download_result

def download_file(self, url, file_destination):
def download_file(self, url, file_destination, timeout=_UNSET):
os.makedirs(os.path.dirname(file_destination), exist_ok=True)
response = requests.get(url)
response = self._request("get", url, timeout=timeout)
if response.status_code // 100 != 2:
raise HTTPError(
f"Error downloading file: {response.status_code} {response.text}"
Expand Down