Skip to content

FloweryAPI

Classes:

Name Description
FloweryAPI

Main class for interacting with the Flowery API.

FloweryAPI

Main class for interacting with the Flowery API.

Example
from pyflowery import FloweryAPI, FloweryAPIConfig

config = FloweryAPIConfig(
    user_agent = "PyFloweryDocumentation/1.0.0",
    token = "hello world",
    allow_truncation = False,
)

api = FloweryAPI(config=config)

Methods:

Name Description
fetch_tts

Fetch a TTS audio file from the Flowery API

fetch_voices

Fetch a list of voices from the Flowery API.

get_voice

Get a voice from the cache using its ID.

get_voices

Get a filtered set of voices from the cache.

populate_voices_cache

Populates the voices cache.

Attributes:

Name Type Description
adapter RestAdapter

Rest Adapter used for making HTTP requests.

config FloweryAPIConfig

Object for configuring requests to the Flowery API.

Source code in pyflowery/pyflowery.py
class FloweryAPI:
    """Main class for interacting with the Flowery API.

    Example:
        ```python
        from pyflowery import FloweryAPI, FloweryAPIConfig

        config = FloweryAPIConfig(
            user_agent = "PyFloweryDocumentation/1.0.0",
            token = "hello world",
            allow_truncation = False,
        )

        api = FloweryAPI(config=config)
        ```
    """

    def __init__(self, config: models.FloweryAPIConfig) -> None:
        self.config: models.FloweryAPIConfig = config
        "Object for configuring requests to the Flowery API."
        self.adapter: rest_adapter.RestAdapter = rest_adapter.RestAdapter(config=config)
        "Rest Adapter used for making HTTP requests."

        self._voices_cache: dict[str, models.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:
        """Populates the voices cache.

        Warning:
            This method is called automatically when the FloweryAPI object is created, and should only be called directly if you need to refresh the voices cache.
        """
        self._voices_cache = {voice.id.lower(): voice async for voice in self.fetch_voices()}
        if not self._voices_cache:
            raise ValueError("Failed to populate voices cache!")
        self.config.logger.info("Voices cache populated with %d entries!", len(self._voices_cache))

    def get_voice(self, voice_id: str) -> models.Voice | None:
        """Get a voice from the cache using its ID.

        Example:
            ```python
            from pyflowery import FloweryAPI, FloweryAPIConfig

            api = FloweryAPI(config=FloweryAPIConfig(user_agent="PyFloweryDocumentation/1.0.0"))
            voice = api.get_voice("372a5e97-1645-563a-9097-36bd83184ab4")
            # Voice(
            #   id='372a5e97-1645-563a-9097-36bd83184ab4',
            #   name='Xiaoyi', gender='Female', source='Microsoft Azure',
            #   language=Language(name='Chinese (China)', code='zh-CN')
            # )
            ```

        Args:
            voice_id (str): The ID of the voice to retrieve from the cache.

        Returns:
            The matching [`Voice`][models.Voice] if found, otherwise `None`.
        """
        return self._voices_cache.get(voice_id.lower())

    def get_voices(
        self, name: str | None = None, gender: str | None = None, source: str | None = None, language: models.Language | None = None
    ) -> tuple[models.Voice, ...] | None:
        """Get a filtered set of voices from the cache.

        Example:
            ```python
            from pyflowery import FloweryAPI, FloweryAPIConfig, Language

            api = FloweryAPI(config=FloweryAPIConfig(user_agent="PyFloweryDocumentation/1.0.0"))
            voices = api.get_voices(source="TikTok", language=Language(name="English (United States)", code="en-US"))
            # (
            #   Voice(
            #       id='fa3ea565-121f-5efd-b4e9-59895c77df23',
            #       name='Alexander', gender='Male', source='TikTok',
            #       language=Language(name='English (United States)', code='en-US')
            #   ),
            #   Voice(
            #       id='a55b0ad0-84c8-597d-832b-0bc4c8777054',
            #       name='Alto', gender='Female', source='TikTok',
            #       language=Language(name='English (United States)', code='en-US')
            #   ), ...
            ```

        Args:
            name: The name to filter results by.
            gender: The gender to filter results by.
            source: The source to filter results by.
            language: The language to filter results by.

        Returns:
            A tuple of [`Voice`][models.Voice] objects if found, otherwise `None`.
            If no keyword arguments are provided, the contents of the internal voice cache will be converted to a tuple and returned instead.
        """
        if not any([name, gender, source, language]):
            return tuple(self._voices_cache.values()) or None

        voices: list[models.Voice] = []

        for voice in self._voices_cache.values():
            if name is not None and voice.name.lower() != name.lower():
                continue
            if gender is not None and voice.gender.lower() != gender.lower():
                continue
            if source is not None and voice.source.lower() != source.lower():
                continue
            if language is not None and voice.language != language:
                continue
            voices.append(voice)

        return tuple(voices) if voices else None

    async def fetch_voices(self) -> AsyncGenerator[models.Voice, None]:
        """Fetch a list of voices from the Flowery API.

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

        Returns:
            A generator of Voices.
        """
        request = await self.adapter.get(endpoint="/tts/voices")
        if request is None:
            raise exceptions.ResponseError("Invalid response from Flowery API: Empty Response!}")
        if request.success is not True or not isinstance(request.data, dict):
            raise exceptions.ResponseError(f"Invalid response from Flowery API: {request.data!r}")

        data: list[dict[str, str]] = request.data.get("voices", [])  # pyright: ignore[reportAny]
        for voice in data:
            try:
                v = models.Voice.model_validate(voice)
            except ValidationError as e:
                self.config.logger.error("Failed to validate voice data. Voice Data: %s, Error: %s", voice, str(e), exc_info=True)
                continue
            yield v

    async def fetch_tts(
        self,
        text: str,
        voice: models.Voice | None = None,
        translate: bool = False,
        silence: timedelta | int | None = None,
        audio_format: models.AudioFormat = models.AudioFormat.MP3,
        speed: float = 1.0,
    ) -> models.TTSResponse:
        """Fetch a TTS audio file from the Flowery API

        Args:
            text (str): The text to convert to speech.
            voice (models.Voice): The voice to use for the speech.
            translate (bool): Whether to translate the text.
            silence (timedelta | int | None): Number of seconds of silence to add to the end of the audio.
            audio_format (models.AudioFormat): 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`][models.FloweryAPIConfig] class is set to `False` (default).
            exceptions.TooManyRequests: Raised when the Flowery API returns a 429 status code.
            exceptions.ClientError: Raised when the Flowery API returns a 4xx status code.
            exceptions.InternalServerError: Raised when the Flowery API returns a 5xx status code.
            exceptions.ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format.
            exceptions.RetryLimitExceeded: Raised when the retry limit defined in the [`FloweryAPIConfig`][models.FloweryAPIConfig] class (default 3) is exceeded.

        Returns:
            An object containing the parameters used to synthesize the text, as well as the raw tts data 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")

        silence = models.TTSResponse.silence_validator(silence)
        speed = models.TTSResponse.speed_validator(speed)

        params = {
            "text": text,
            "translate": str(translate).lower(),
            "silence": ceil(silence.total_seconds() * 1000),  # milliseconds
            "audio_format": audio_format.value,
            "speed": speed,
        }
        if voice:
            params["voice"] = voice.id

        request = await self.adapter.get(endpoint="/tts", params=params, timeout=180)

        if request is not None:
            if isinstance(request.data, bytes):
                return models.TTSResponse(
                    data=request.data, text=text, translate=translate, voice=voice, silence=silence, audio_format=audio_format, speed=speed
                )
            raise exceptions.ResponseError(f"Invalid response from Flowery API: {request.data!r}")
        raise exceptions.ResponseError("Invalid response from Flowery API: Empty Response!}")

adapter = rest_adapter.RestAdapter(config=config) instance-attribute

Rest Adapter used for making HTTP requests.

config = config instance-attribute

Object for configuring requests to the Flowery API.

fetch_tts(text, voice=None, translate=False, silence=None, audio_format=models.AudioFormat.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

The voice to use for the speech.

None
translate bool

Whether to translate the text.

False
silence timedelta | int | None

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

None
audio_format AudioFormat

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:

Type Description
TTSResponse

An object containing the parameters used to synthesize the text, as well as the raw tts data in bytes.

Source code in pyflowery/pyflowery.py
async def fetch_tts(
    self,
    text: str,
    voice: models.Voice | None = None,
    translate: bool = False,
    silence: timedelta | int | None = None,
    audio_format: models.AudioFormat = models.AudioFormat.MP3,
    speed: float = 1.0,
) -> models.TTSResponse:
    """Fetch a TTS audio file from the Flowery API

    Args:
        text (str): The text to convert to speech.
        voice (models.Voice): The voice to use for the speech.
        translate (bool): Whether to translate the text.
        silence (timedelta | int | None): Number of seconds of silence to add to the end of the audio.
        audio_format (models.AudioFormat): 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`][models.FloweryAPIConfig] class is set to `False` (default).
        exceptions.TooManyRequests: Raised when the Flowery API returns a 429 status code.
        exceptions.ClientError: Raised when the Flowery API returns a 4xx status code.
        exceptions.InternalServerError: Raised when the Flowery API returns a 5xx status code.
        exceptions.ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format.
        exceptions.RetryLimitExceeded: Raised when the retry limit defined in the [`FloweryAPIConfig`][models.FloweryAPIConfig] class (default 3) is exceeded.

    Returns:
        An object containing the parameters used to synthesize the text, as well as the raw tts data 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")

    silence = models.TTSResponse.silence_validator(silence)
    speed = models.TTSResponse.speed_validator(speed)

    params = {
        "text": text,
        "translate": str(translate).lower(),
        "silence": ceil(silence.total_seconds() * 1000),  # milliseconds
        "audio_format": audio_format.value,
        "speed": speed,
    }
    if voice:
        params["voice"] = voice.id

    request = await self.adapter.get(endpoint="/tts", params=params, timeout=180)

    if request is not None:
        if isinstance(request.data, bytes):
            return models.TTSResponse(
                data=request.data, text=text, translate=translate, voice=voice, silence=silence, audio_format=audio_format, speed=speed
            )
        raise exceptions.ResponseError(f"Invalid response from Flowery API: {request.data!r}")
    raise exceptions.ResponseError("Invalid response from Flowery API: Empty Response!}")

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]

A generator of Voices.

Source code in pyflowery/pyflowery.py
async def fetch_voices(self) -> AsyncGenerator[models.Voice, None]:
    """Fetch a list of voices from the Flowery API.

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

    Returns:
        A generator of Voices.
    """
    request = await self.adapter.get(endpoint="/tts/voices")
    if request is None:
        raise exceptions.ResponseError("Invalid response from Flowery API: Empty Response!}")
    if request.success is not True or not isinstance(request.data, dict):
        raise exceptions.ResponseError(f"Invalid response from Flowery API: {request.data!r}")

    data: list[dict[str, str]] = request.data.get("voices", [])  # pyright: ignore[reportAny]
    for voice in data:
        try:
            v = models.Voice.model_validate(voice)
        except ValidationError as e:
            self.config.logger.error("Failed to validate voice data. Voice Data: %s, Error: %s", voice, str(e), exc_info=True)
            continue
        yield v

get_voice(voice_id)

Get a voice from the cache using its ID.

Example
from pyflowery import FloweryAPI, FloweryAPIConfig

api = FloweryAPI(config=FloweryAPIConfig(user_agent="PyFloweryDocumentation/1.0.0"))
voice = api.get_voice("372a5e97-1645-563a-9097-36bd83184ab4")
# Voice(
#   id='372a5e97-1645-563a-9097-36bd83184ab4',
#   name='Xiaoyi', gender='Female', source='Microsoft Azure',
#   language=Language(name='Chinese (China)', code='zh-CN')
# )

Parameters:

Name Type Description Default
voice_id str

The ID of the voice to retrieve from the cache.

required

Returns:

Type Description
Voice | None

The matching Voice if found, otherwise None.

Source code in pyflowery/pyflowery.py
def get_voice(self, voice_id: str) -> models.Voice | None:
    """Get a voice from the cache using its ID.

    Example:
        ```python
        from pyflowery import FloweryAPI, FloweryAPIConfig

        api = FloweryAPI(config=FloweryAPIConfig(user_agent="PyFloweryDocumentation/1.0.0"))
        voice = api.get_voice("372a5e97-1645-563a-9097-36bd83184ab4")
        # Voice(
        #   id='372a5e97-1645-563a-9097-36bd83184ab4',
        #   name='Xiaoyi', gender='Female', source='Microsoft Azure',
        #   language=Language(name='Chinese (China)', code='zh-CN')
        # )
        ```

    Args:
        voice_id (str): The ID of the voice to retrieve from the cache.

    Returns:
        The matching [`Voice`][models.Voice] if found, otherwise `None`.
    """
    return self._voices_cache.get(voice_id.lower())

get_voices(name=None, gender=None, source=None, language=None)

Get a filtered set of voices from the cache.

Example
from pyflowery import FloweryAPI, FloweryAPIConfig, Language

api = FloweryAPI(config=FloweryAPIConfig(user_agent="PyFloweryDocumentation/1.0.0"))
voices = api.get_voices(source="TikTok", language=Language(name="English (United States)", code="en-US"))
# (
#   Voice(
#       id='fa3ea565-121f-5efd-b4e9-59895c77df23',
#       name='Alexander', gender='Male', source='TikTok',
#       language=Language(name='English (United States)', code='en-US')
#   ),
#   Voice(
#       id='a55b0ad0-84c8-597d-832b-0bc4c8777054',
#       name='Alto', gender='Female', source='TikTok',
#       language=Language(name='English (United States)', code='en-US')
#   ), ...

Parameters:

Name Type Description Default
name str | None

The name to filter results by.

None
gender str | None

The gender to filter results by.

None
source str | None

The source to filter results by.

None
language Language | None

The language to filter results by.

None

Returns:

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

A tuple of Voice objects if found, otherwise None.

tuple[Voice, ...] | None

If no keyword arguments are provided, the contents of the internal voice cache will be converted to a tuple and returned instead.

Source code in pyflowery/pyflowery.py
def get_voices(
    self, name: str | None = None, gender: str | None = None, source: str | None = None, language: models.Language | None = None
) -> tuple[models.Voice, ...] | None:
    """Get a filtered set of voices from the cache.

    Example:
        ```python
        from pyflowery import FloweryAPI, FloweryAPIConfig, Language

        api = FloweryAPI(config=FloweryAPIConfig(user_agent="PyFloweryDocumentation/1.0.0"))
        voices = api.get_voices(source="TikTok", language=Language(name="English (United States)", code="en-US"))
        # (
        #   Voice(
        #       id='fa3ea565-121f-5efd-b4e9-59895c77df23',
        #       name='Alexander', gender='Male', source='TikTok',
        #       language=Language(name='English (United States)', code='en-US')
        #   ),
        #   Voice(
        #       id='a55b0ad0-84c8-597d-832b-0bc4c8777054',
        #       name='Alto', gender='Female', source='TikTok',
        #       language=Language(name='English (United States)', code='en-US')
        #   ), ...
        ```

    Args:
        name: The name to filter results by.
        gender: The gender to filter results by.
        source: The source to filter results by.
        language: The language to filter results by.

    Returns:
        A tuple of [`Voice`][models.Voice] objects if found, otherwise `None`.
        If no keyword arguments are provided, the contents of the internal voice cache will be converted to a tuple and returned instead.
    """
    if not any([name, gender, source, language]):
        return tuple(self._voices_cache.values()) or None

    voices: list[models.Voice] = []

    for voice in self._voices_cache.values():
        if name is not None and voice.name.lower() != name.lower():
            continue
        if gender is not None and voice.gender.lower() != gender.lower():
            continue
        if source is not None and voice.source.lower() != source.lower():
            continue
        if language is not None and voice.language != language:
            continue
        voices.append(voice)

    return tuple(voices) if voices else None

populate_voices_cache() async

Populates the voices cache.

Warning

This method is called automatically when the FloweryAPI object is created, and should only be called directly if you need to refresh the voices cache.

Source code in pyflowery/pyflowery.py
async def populate_voices_cache(self) -> None:
    """Populates the voices cache.

    Warning:
        This method is called automatically when the FloweryAPI object is created, and should only be called directly if you need to refresh the voices cache.
    """
    self._voices_cache = {voice.id.lower(): voice async for voice in self.fetch_voices()}
    if not self._voices_cache:
        raise ValueError("Failed to populate voices cache!")
    self.config.logger.info("Voices cache populated with %d entries!", len(self._voices_cache))