Merge pull request #473 from dvonthenen/fix-manaage-null-response

Fix `get_usage_request` when Response Not Available

deepgram/clients/manage/v1/async_client.py

@@ -4,12 +4,13 @@
 
 import logging
 from typing import Dict, Union, Optional
+import json
 
 import httpx
 
 from ....utils import verboselogs
 from ....options import DeepgramClientOptions
-from ...common import AbstractAsyncRestClient
+from ...common import AbstractAsyncRestClient, DeepgramError
 
 from .response import (
     Message,
@@ -1000,6 +1001,14 @@ async def get_usage_request(
             url, timeout=timeout, addons=addons, headers=headers, **kwargs
         )
         self._logger.info("result: %s", result)
+
+        # convert str to JSON to check response field
+        json_result = json.loads(result)
+        if json_result.get("response") is None:
+            raise DeepgramError(
+                "Response is not available yet. Please try again later."
+            )
+
         res = UsageRequest.from_json(result)
         self._logger.verbose("result: %s", res)
         self._logger.notice("get_usage_request succeeded")

deepgram/clients/manage/v1/client.py

@@ -4,12 +4,13 @@
 
 import logging
 from typing import Dict, Union, Optional
+import json
 
 import httpx
 
 from ....utils import verboselogs
 from ....options import DeepgramClientOptions
-from ...common import AbstractSyncRestClient
+from ...common import AbstractSyncRestClient, DeepgramError
 
 from .response import (
     Message,
@@ -1003,6 +1004,14 @@ def get_usage_request(
             url, timeout=timeout, addons=addons, headers=headers, **kwargs
         )
         self._logger.info("json: %s", result)
+
+        # convert str to JSON to check response field
+        json_result = json.loads(result)
+        if json_result.get("response") is None:
+            raise DeepgramError(
+                "Response is not available yet. Please try again later."
+            )
+
         res = UsageRequest.from_json(result)
         self._logger.verbose("result: %s", res)
         self._logger.notice("get_usage_request succeeded")

tests/edge_cases/usage_to_fast/main.py

@@ -0,0 +1,83 @@
+# Copyright 2024 Deepgram SDK contributors. All Rights Reserved.
+# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
+# SPDX-License-Identifier: MIT
+
+import os
+import sys
+from dotenv import load_dotenv
+import logging
+from deepgram.utils import verboselogs
+
+import httpx
+
+from deepgram import (
+    DeepgramClient,
+    DeepgramClientOptions,
+    PrerecordedOptions,
+    FileSource,
+)
+
+load_dotenv()
+
+AUDIO_FILE = "sample.mp3"
+
+
+def main():
+    try:
+        # Create a Deepgram client using the API key
+        # config: DeepgramClientOptions = DeepgramClientOptions(verbose=verboselogs.SPAM)
+        config: DeepgramClientOptions = DeepgramClientOptions()
+        deepgram: DeepgramClient = DeepgramClient("", config)
+
+        # get projects
+        projectResp = deepgram.manage.v("1").get_projects()
+        if projectResp is None:
+            print(f"ListProjects failed.")
+            sys.exit(1)
+
+        myId = None
+        myName = None
+        for project in projectResp.projects:
+            myId = project.project_id
+            myName = project.name
+            print(f"ListProjects() - ID: {myId}, Name: {myName}")
+            break
+
+        with open(AUDIO_FILE, "rb") as file:
+            buffer_data = file.read()
+
+        payload: FileSource = {
+            "buffer": buffer_data,
+        }
+
+        options: PrerecordedOptions = PrerecordedOptions(
+            callback="http://example.com",
+            model="nova-2",
+            smart_format=True,
+            utterances=True,
+            punctuate=True,
+            diarize=True,
+        )
+
+        response = deepgram.listen.rest.v("1").transcribe_file(
+            payload, options, timeout=httpx.Timeout(300.0, connect=10.0)
+        )
+        request_id = (
+            response.request_id
+        )  # without callback: response.metadata.request_id
+        print(f"request_id: {request_id}")
+
+        # get request
+        getResp = deepgram.manage.v("1").get_usage_request(myId, request_id)
+        if getResp is None:
+            print("No request found")
+        else:
+            print(f"GetUsageRequest() - getResp: {getResp}")
+        print("\n\n\n")
+
+    except Exception as e:
+        print(f"Exception: {e}")
+
+
+if __name__ == "__main__":
+    main()