Skip to content

Commit

Permalink
Refactor CambAI session and API URL to be private
Browse files Browse the repository at this point in the history
  • Loading branch information
@cr2007
cr2007 committed Jul 9, 2024
1 parent 2a58873 commit d97f949
Showing 1 changed file with 21 additions and 21 deletions.
42 changes: 21 additions & 21 deletions cambai/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -288,13 +288,13 @@ def __init__(self, *, api_key: Optional[str] = None) -> None:
)

# Set the base URL for the Camb AI API
self.camb_api_url: str = "https://client.camb.ai/apis/"
self.__camb_api_url: str = "https://client.camb.ai/apis/"

# Create a new session for making requests
self.session: requests.Session = requests.Session()
self.__session: requests.Session = requests.Session()

# Set the API key in the session headers
self.session.headers = {"x-api-key": api_key}
self.__session.headers = {"x-api-key": api_key}


def __create_api_endpoint(self, /, endpoint: str) -> str:
Expand All @@ -309,7 +309,7 @@ def __create_api_endpoint(self, /, endpoint: str) -> str:
"""

# Append the provided endpoint to the base URL
return self.camb_api_url + endpoint
return self.__camb_api_url + endpoint


def get_languages(
Expand Down Expand Up @@ -341,7 +341,7 @@ def get_languages(
url: str = self.__create_api_endpoint(f"{language_type}_languages")

# Send a GET request to the constructed URL
response: requests.Response = self.session.get(url)
response: requests.Response = self.__session.get(url)

# Check for a successful response (status code 200)
if response.status_code != 200:
Expand Down Expand Up @@ -423,7 +423,7 @@ def create_custom_voice(
try:
# Open the file in binary read mode and send the POST request
with open(file, "rb") as file_resource:
response: requests.Response = self.session.post(
response: requests.Response = self.__session.post(
url=url, files={"file": file_resource}, data=data
)

Expand Down Expand Up @@ -485,7 +485,7 @@ def get_all_voices(
url: str = self.__create_api_endpoint("list_voices")

# Send a GET request to the constructed URL
response: requests.Response = self.session.get(url)
response: requests.Response = self.__session.get(url)

# Check for a successful response (status code 200)
if response.status_code != 200:
Expand Down Expand Up @@ -562,7 +562,7 @@ def get_task_status(
)

# Send a GET request to the API endpoint
response: requests.Response = self.session.get(url)
response: requests.Response = self.__session.get(url)

# Check for a successful response (status code 200)
if response.status_code != 200:
Expand Down Expand Up @@ -605,7 +605,7 @@ def start_dubbing(
url: str = self.__create_api_endpoint("end_to_end_dubbing")

# Set the content type for the request to JSON
self.session.headers["Content-Type"] = "application/json"
self.__session.headers["Content-Type"] = "application/json"

# Prepare the data payload for the POST request
data: dict = {
Expand All @@ -615,7 +615,7 @@ def start_dubbing(
}

# Send a POST request with the video URL and language IDs
response: requests.Response = self.session.post(url=url, json=data)
response: requests.Response = self.__session.post(url=url, json=data)

# Check for a successful response (status code 200)
if response.status_code != 200:
Expand Down Expand Up @@ -671,7 +671,7 @@ def get_dubbed_run_info(self, /, run_id: int) -> DubbedRunInfo:
url: str = self.__create_api_endpoint(f"dubbed_run_info/{run_id}")

# Send a GET request to the API endpoint
response: requests.Response = self.session.get(url)
response: requests.Response = self.__session.get(url)

# Check for a successful response (status code 200)
if response.status_code != 200:
Expand Down Expand Up @@ -853,10 +853,10 @@ def create_tts(
data["age"] = age

# Set the request content type to JSON
self.session.headers["Content-Type"] = "application/json"
self.__session.headers["Content-Type"] = "application/json"

# Execute the POST request
response: requests.Response = self.session.post(url=url, json=data)
response: requests.Response = self.__session.post(url=url, json=data)

# Handle unsuccessful request
if response.status_code != 200:
Expand Down Expand Up @@ -910,7 +910,7 @@ def get_tts_result(
url: str = self.__create_api_endpoint(f"tts_result/{run_id}")

# Send a GET request to the API endpoint
response: requests.Response = self.session.get(url, stream=True)
response: requests.Response = self.__session.get(url, stream=True)

# Check for a successful response (status code 200)
if response.status_code != 200:
Expand Down Expand Up @@ -1085,7 +1085,7 @@ def create_transcription(
# Open the audio file in binary read mode
with open(audio_file, "rb") as audio:
# Make a POST request to the API with the audio file and language ID
response: requests.Response = self.session.post(
response: requests.Response = self.__session.post(
url=url,
files={"file": audio}, # The audio file to be transcribed
data=data, # Additional data including the language ID
Expand Down Expand Up @@ -1159,7 +1159,7 @@ def get_transcription_result(
url: str = self.__create_api_endpoint(f"transcription_result/{run_id}")

# Perform a GET request to the API endpoint
response: requests.Response = self.session.get(url)
response: requests.Response = self.__session.get(url)

# Check if the response status code indicates a successful request
if response.status_code != 200:
Expand Down Expand Up @@ -1346,7 +1346,7 @@ def create_translation(
url: str = self.__create_api_endpoint("create_translation")

# Set the content type for the request
self.session.headers["Content-Type"] = "application/json"
self.__session.headers["Content-Type"] = "application/json"

# Construct the data payload for the POST request
data: ExtendedTranslationData = {
Expand All @@ -1364,7 +1364,7 @@ def create_translation(
data["gender"] = gender.value

# Send the POST request to the API
response: requests.Response = self.session.post(url=url, json=data)
response: requests.Response = self.__session.post(url=url, json=data)

# Check for successful response
if response.status_code != 200:
Expand Down Expand Up @@ -1414,7 +1414,7 @@ def get_translation_result(
url: str = self.__create_api_endpoint(f"translation_result/{run_id}")

# Send a GET request to the API endpoint and get the response
response: requests.Response = self.session.get(url)
response: requests.Response = self.__session.get(url)

if response.status_code != 200:
print(f"Error: There was a {response.status_code} error with your GET request.")
Expand Down Expand Up @@ -1603,7 +1603,7 @@ def create_translated_tts(
url: str = self.__create_api_endpoint("create_translated_tts")

# Set the content type for the request
self.session.headers["Content-Type"] = "application/json"
self.__session.headers["Content-Type"] = "application/json"

# Prepare the data payload for the POST request
data: ExtendedTranslationTTSData = {
Expand All @@ -1622,7 +1622,7 @@ def create_translated_tts(
data["gender"] = gender

# Make the POST request to the API endpoint
response: requests.Response = self.session.post(url=url, json=data)
response: requests.Response = self.__session.post(url=url, json=data)

# Check for successful response
if response.status_code != 200:
Expand Down

0 comments on commit d97f949

Please sign in to comment.