Skip to content

FloweryAPI

FloweryAPI

Main class for interacting with the Flowery API

Attributes:

Name Type Description
config FloweryAPIConfig

Configuration object for the API
adapter RestAdapter

Adapter for making HTTP requests
Source code in pyflowery/pyflowery.py 
  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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class FloweryAPI:
    """Main class for interacting with the Flowery API

    Attributes:
        config (FloweryAPIConfig): Configuration object for the API
        adapter (RestAdapter): Adapter for making HTTP requests
    """

    def __init__(self, config: FloweryAPIConfig) -> None:
        self.config = config
        self.adapter = RestAdapter(config=config)
        self._voices_cache: Tuple[Voice, ...] = ()
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop and loop.is_running():
            self.config.logger.info("Async event loop is already running. Adding `_populate_voices_cache()` to the event loop.")
            asyncio.create_task(self._populate_voices_cache())
        else:
            asyncio.run(self._populate_voices_cache())

    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
        if not self._voices_cache:
            raise ValueError("Failed to populate voices cache! Please report this issue at https://www.coastalcommits.com/cswimr/PyFlowery/issues.")
        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.

        Args:
            voice_id (str): The ID of the voice
            name (str): The name of the voice

        Returns:
            Tuple[Voice] | None: A tuple of Voice objects if found, otherwise None
        """
        if voice_id:
            voice = next((voice for voice in self._voices_cache if voice.id == voice_id), None)
            return (voice,) if voice else None
        if name:
            voices = []
            for voice in self._voices_cache:
                if voice.name == name:
                    voices.append(voice)
            return tuple(voices) or None
        return self._voices_cache or None

    async def fetch_voice(self, voice_id: str) -> Voice:
        """Fetch a voice from the Flowery API. This method bypasses the cache and directly queries the Flowery API. You should usually use `get_voices()` instead.

        Args:
            voice_id (str): The ID of the voice

        Raises:
            ValueError: Raised when the voice is not found
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            Voice: The voice
        """
        async for voice in self.fetch_voices():
            if voice.id == voice_id:
                return voice
        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

        Raises:
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            AsyncGenerator[Voice, None]: A generator of Voices
        """
        request = await self.adapter.get(endpoint="/tts/voices")
        if request is not None:
            if isinstance(request.data, dict):
                for voice in request.data.get("voices", []):
                    yield Voice(
                        id=voice["id"],
                        name=voice["name"],
                        gender=voice["gender"],
                        source=voice["source"],
                        language=Language(**voice["language"]),
                    )
            else:
                raise ResponseError(f"Invalid response from Flowery API: {request.data!r}")
        else:
            raise ResponseError("Invalid response from Flowery API: Empty Response!}")

    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:
            text (str): The text to convert to speech
            voice (Voice | str): The voice to use for the speech
            translate (bool): Whether to translate the text
            silence (int): Number of seconds of silence to add to the end of the audio
            audio_format (str): The audio format to return
            speed (float): The speed of the speech

        Raises:
            ValueError: Raised when the provided text is too long and `allow_truncation` in the `FloweryAPIConfig` class is set to `False` (default).
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            bytes: The audio file in bytes
        """
        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")
        params = {
            "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(endpoint="/tts", params=params, timeout=180)

        if request is not None:
            if isinstance(request.data, bytes):
                return request.data
            raise ResponseError(f"Invalid response from Flowery API: {request.data!r}")
        raise ResponseError("Invalid response from Flowery API: Empty Response!}")

fetch_tts(text, voice=None, translate=False, silence=0, audio_format='mp3', speed=1.0) async

Fetch a TTS audio file from the Flowery API

Parameters:

Name Type Description Default
text str

The text to convert to speech

 required
voice Voice | str

The voice to use for the speech

 None
translate bool

Whether to translate the text

 False
silence int

Number of seconds of silence to add to the end of the audio

 0
audio_format str

The audio format to return

 'mp3'
speed float

The speed of the speech

 1.0

Raises:

Type Description
ValueError

Raised when the provided text is too long and allow_truncation in the FloweryAPIConfig class is set to False (default).
TooManyRequests

Raised when the Flowery API returns a 429 status code
ClientError

Raised when the Flowery API returns a 4xx status code
InternalServerError

Raised when the Flowery API returns a 5xx status code
ResponseError

Raised when the Flowery API returns an empty response or a response with an unexpected format
RetryLimitExceeded

Raised when the retry limit defined in the FloweryAPIConfig class (default 3) is exceeded

Returns:

Name Type Description
bytes bytes

The audio file in bytes
Source code in pyflowery/pyflowery.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
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:
        text (str): The text to convert to speech
        voice (Voice | str): The voice to use for the speech
        translate (bool): Whether to translate the text
        silence (int): Number of seconds of silence to add to the end of the audio
        audio_format (str): The audio format to return
        speed (float): The speed of the speech

    Raises:
        ValueError: Raised when the provided text is too long and `allow_truncation` in the `FloweryAPIConfig` class is set to `False` (default).
        TooManyRequests: Raised when the Flowery API returns a 429 status code
        ClientError: Raised when the Flowery API returns a 4xx status code
        InternalServerError: Raised when the Flowery API returns a 5xx status code
        ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format
        RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

    Returns:
        bytes: The audio file in bytes
    """
    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")
    params = {
        "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(endpoint="/tts", params=params, timeout=180)

    if request is not None:
        if isinstance(request.data, bytes):
            return request.data
        raise ResponseError(f"Invalid response from Flowery API: {request.data!r}")
    raise ResponseError("Invalid response from Flowery API: Empty Response!}")

fetch_voice(voice_id) async

Fetch a voice from the Flowery API. This method bypasses the cache and directly queries the Flowery API. You should usually use get_voices() instead.

Parameters:

Name Type Description Default
voice_id str

The ID of the voice

 required

Raises:

Type Description
ValueError

Raised when the voice is not found
TooManyRequests

Raised when the Flowery API returns a 429 status code
ClientError

Raised when the Flowery API returns a 4xx status code
InternalServerError

Raised when the Flowery API returns a 5xx status code
RetryLimitExceeded

Raised when the retry limit defined in the FloweryAPIConfig class (default 3) is exceeded

Returns:

Name Type Description
Voice Voice

The voice
Source code in pyflowery/pyflowery.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def fetch_voice(self, voice_id: str) -> Voice:
    """Fetch a voice from the Flowery API. This method bypasses the cache and directly queries the Flowery API. You should usually use `get_voices()` instead.

    Args:
        voice_id (str): The ID of the voice

    Raises:
        ValueError: Raised when the voice is not found
        TooManyRequests: Raised when the Flowery API returns a 429 status code
        ClientError: Raised when the Flowery API returns a 4xx status code
        InternalServerError: Raised when the Flowery API returns a 5xx status code
        RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

    Returns:
        Voice: The voice
    """
    async for voice in self.fetch_voices():
        if voice.id == voice_id:
            return voice
    raise ValueError(f"Voice with ID {voice_id} not found.")

fetch_voices() async

Fetch a list of voices from the Flowery API

Raises:

Type Description
TooManyRequests

Raised when the Flowery API returns a 429 status code
ClientError

Raised when the Flowery API returns a 4xx status code
InternalServerError

Raised when the Flowery API returns a 5xx status code
ResponseError

Raised when the Flowery API returns an empty response or a response with an unexpected format
RetryLimitExceeded

Raised when the retry limit defined in the FloweryAPIConfig class (default 3) is exceeded

Returns:

Type Description
AsyncGenerator[Voice, None]

AsyncGenerator[Voice, None]: A generator of Voices
Source code in pyflowery/pyflowery.py 
 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
async def fetch_voices(self) -> AsyncGenerator[Voice, None]:
    """Fetch a list of voices from the Flowery API

    Raises:
        TooManyRequests: Raised when the Flowery API returns a 429 status code
        ClientError: Raised when the Flowery API returns a 4xx status code
        InternalServerError: Raised when the Flowery API returns a 5xx status code
        ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format
        RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

    Returns:
        AsyncGenerator[Voice, None]: A generator of Voices
    """
    request = await self.adapter.get(endpoint="/tts/voices")
    if request is not None:
        if isinstance(request.data, dict):
            for voice in request.data.get("voices", []):
                yield Voice(
                    id=voice["id"],
                    name=voice["name"],
                    gender=voice["gender"],
                    source=voice["source"],
                    language=Language(**voice["language"]),
                )
        else:
            raise ResponseError(f"Invalid response from Flowery API: {request.data!r}")
    else:
        raise ResponseError("Invalid response from Flowery API: Empty Response!}")

get_voices(voice_id=None, name=None)

Get a set of voices from the cache.

Parameters:

Name Type Description Default
voice_id str

The ID of the voice

 None
name str

The name of the voice

 None

Returns:

Type Description
Tuple[Voice, ...] | None

Tuple[Voice] | None: A tuple of Voice objects if found, otherwise None
Source code in pyflowery/pyflowery.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_voices(self, voice_id: str | None = None, name: str | None = None) -> Tuple[Voice, ...] | None:
    """Get a set of voices from the cache.

    Args:
        voice_id (str): The ID of the voice
        name (str): The name of the voice

    Returns:
        Tuple[Voice] | None: A tuple of Voice objects if found, otherwise None
    """
    if voice_id:
        voice = next((voice for voice in self._voices_cache if voice.id == voice_id), None)
        return (voice,) if voice else None
    if name:
        voices = []
        for voice in self._voices_cache:
            if voice.name == name:
                voices.append(voice)
        return tuple(voices) or None
    return self._voices_cache or None