(3.0.0) cleaned up uv dependencies for pep 735, made a backwards incompatible change in the FloweryApiConfig model, and set up ruff formatting
Some checks failed
Actions / Build (push) Successful in 10s
Actions / Lint with Ruff & Pylint (push) Failing after 12s
Actions / Build Documentation (push) Failing after 6s

This commit is contained in:
cswimr 2024-11-15 00:10:38 -05:00
parent dcb5365fea
commit fde5dad155
Signed by: cswimr
GPG key ID: A9C162E867C851FA
11 changed files with 273 additions and 243 deletions

View file

@ -17,11 +17,13 @@ class Voice:
source (str): Source of the voice
language (Language): Language object
"""
id: str
name: str
gender: str
source: str
language: 'Language'
language: "Language"
@dataclass
class Language:
@ -31,9 +33,11 @@ class Language:
name (str): Name of the language
code (str): Code of the language
"""
name: str
code: str
@dataclass
class Result:
"""Result returned from low-level RestAdapter
@ -44,9 +48,10 @@ class Result:
message (str = ''): Human readable result
data (Union[List[Dict], Dict, bytes]): Python List of Dictionaries (or maybe just a single Dictionary on error), can also be a ByteString
"""
success: bool
status_code: int
message: str = ''
message: str = ""
data: Union[List[Dict], Dict, bytes] = field(default_factory=dict)
@ -60,13 +65,18 @@ class FloweryAPIConfig:
allow_truncation (bool): Whether to allow truncation of text that is too long, defaults to `True`
retry_limit (int): Number of times to retry a request before giving up, defaults to `3`
interval (int): Seconds to wait between each retried request, multiplied by how many attempted requests have been made, defaults to `5`
Properties:
prepended_user_agent (str): The user_agent with the PyFlowery module version prepended and the Python version appended
"""
user_agent: str
logger: Logger = getLogger('pyflowery')
logger: Logger = getLogger("pyflowery")
allow_truncation: bool = False
retry_limit: int = 3
interval: int = 5
@property
def prepended_user_agent(self) -> str:
"""Return the user_agent with the PyFlowery module version prepended"""
return f"PyFlowery/{VERSION} {self.user_agent} (Python {pyversion})"

View file

@ -12,7 +12,8 @@ class FloweryAPI:
config (FloweryAPIConfig): Configuration object for the API
adapter (RestAdapter): Adapter for making HTTP requests
"""
def __init__(self, config: FloweryAPIConfig):
def __init__(self, config: FloweryAPIConfig) -> None:
self.config = config
self.adapter = RestAdapter(config)
self._voices_cache: Tuple[Voice] = ()
@ -26,10 +27,10 @@ class FloweryAPI:
else:
asyncio.run(self._populate_voices_cache())
async def _populate_voices_cache(self):
async def _populate_voices_cache(self) -> None:
"""Populate the voices cache. This method is called automatically when the FloweryAPI object is created, and should not be called directly."""
self._voices_cache = tuple([voice async for voice in self.fetch_voices()]) # pylint: disable=consider-using-generator
self.config.logger.info('Voices cache populated!')
self._voices_cache = tuple([voice async for voice in self.fetch_voices()]) # pylint: disable=consider-using-generator
self.config.logger.info("Voices cache populated!")
def get_voices(self, voice_id: str | None = None, name: str | None = None) -> Tuple[Voice] | None:
"""Get a set of voices from the cache.
@ -71,7 +72,7 @@ class FloweryAPI:
async for voice in self.fetch_voices():
if voice.id == voice_id:
return voice
raise ValueError(f'Voice with ID {voice_id} not found.')
raise ValueError(f"Voice with ID {voice_id} not found.")
async def fetch_voices(self) -> AsyncGenerator[Voice, None]:
"""Fetch a list of voices from the Flowery API
@ -85,17 +86,19 @@ class FloweryAPI:
Returns:
AsyncGenerator[Voice, None]: A generator of Voices
"""
request = await self.adapter.get('/tts/voices')
for voice in request.data['voices']:
request = await self.adapter.get(endpoint="/tts/voices")
for voice in request.data["voices"]:
yield Voice(
id=voice['id'],
name=voice['name'],
gender=voice['gender'],
source=voice['source'],
language=Language(**voice['language']),
id=voice["id"],
name=voice["name"],
gender=voice["gender"],
source=voice["source"],
language=Language(**voice["language"]),
)
async def fetch_tts(self, text: str, voice: Voice | str | None = None, translate: bool = False, silence: int = 0, audio_format: str = 'mp3', speed: float = 1.0) -> bytes:
async def fetch_tts(
self, text: str, voice: Voice | str | None = None, translate: bool = False, silence: int = 0, audio_format: str = "mp3", speed: float = 1.0
) -> bytes:
"""Fetch a TTS audio file from the Flowery API
Args:
@ -118,16 +121,16 @@ class FloweryAPI:
"""
if len(text) > 2048:
if not self.config.allow_truncation:
raise ValueError('Text must be less than or equal to 2048 characters')
self.config.logger.warning('Text is too long, will be truncated to 2048 characters by the API')
raise ValueError("Text must be less than or equal to 2048 characters")
self.config.logger.warning("Text is too long, will be truncated to 2048 characters by the API")
params = {
'text': text,
'translate': str(translate).lower(),
'silence': silence,
'audio_format': audio_format,
'speed': speed,
"text": text,
"translate": str(translate).lower(),
"silence": silence,
"audio_format": audio_format,
"speed": speed,
}
if voice:
params['voice'] = voice.id if isinstance(voice, Voice) else voice
request = await self.adapter.get('/tts', params, timeout=180)
params["voice"] = voice.id if isinstance(voice, Voice) else voice
request = await self.adapter.get(endpoint="/tts", params=params, timeout=180)
return request.data

View file

@ -1,4 +1,4 @@
"""This module contains the RestAdapter class, which is used to make requests to the Flowery API."""""
"""This module contains the RestAdapter class, which is used to make requests to the Flowery API.""" ""
from asyncio import sleep as asleep
from json import JSONDecodeError
@ -22,11 +22,12 @@ class RestAdapter:
Raises:
ValueError: Raised when the keyword arguments passed to the class constructor conflict.
"""
def __init__(self, config = FloweryAPIConfig):
def __init__(self, config=FloweryAPIConfig) -> None:
self._url = "https://api.flowery.pw/v1"
self.config = config
async def _do(self, http_method: str, endpoint: str, params: dict = None, timeout: float = 60):
async def _do(self, http_method: str, endpoint: str, params: dict = None, timeout: float = 60) -> Result | None:
"""Internal method to make a request to the Flowery API. You shouldn't use this directly.
Args:
@ -46,7 +47,7 @@ class RestAdapter:
"""
full_url = self._url + endpoint
headers = {
'User-Agent': self.config.prepended_user_agent(),
"User-Agent": self.config.prepended_user_agent,
}
sanitized_params = {k: str(v) if isinstance(v, bool) else v for k, v in params.items()} if params else None
retry_counter = 0
@ -84,9 +85,11 @@ class RestAdapter:
raise RetryLimitExceeded(message=f"Request failed more than {self.config.retry_limit} times, not retrying") from e
return result
async def get(self, endpoint: str, params: dict = None, timeout: float = 60) -> Result:
async def get(self, endpoint: str, params: dict = None, timeout: float = 60) -> Result | None:
"""Make a GET request to the Flowery API. You should almost never have to use this directly.
If you need to use this method because an endpoint is missing, please open an issue on the [CoastalCommits repository](https://www.coastalcommits.com/cswimr/PyFlowery/issues).
Args:
endpoint (str): The endpoint to make the request to.
params (dict): Python dictionary of query parameters to send with the request.
@ -101,4 +104,4 @@ class RestAdapter:
Returns:
Result: A Result object containing the status code, message, and data from the request.
"""
return await self._do(http_method='GET', endpoint=endpoint, params=params, timeout=timeout)
return await self._do(http_method="GET", endpoint=endpoint, params=params, timeout=timeout)

View file

@ -1 +1 @@
VERSION = "2.1.3"
VERSION = "3.0.0"