openai_utils

Shared utilities for interacting with the OpenAI API.

Classes:

Name	Description
AIModels	The AI models available for use by an OpenAI client, grouped by type.

Functions:

Name	Description
execute_request_with_retry	Calls an async function and retries with exponential backoff on RateLimitError.
get_usable_models_async	Asynchronously get the usable models, fetching them if not already cached.
get_usable_models_sync	Synchronously get the usable models using asyncio.run().

Attributes:

Name	Type	Description
T
tts_voices	list[str]

T module-attribute

T = TypeVar('T')

tts_voices module-attribute

tts_voices: list[str] = extract_literal_values_from_member(
    SpeechCreateParams, "voice"
)

AIModels dataclass

AIModels(
    image_models: list[str],
    speech_models: list[str],
    text_models: list[str],
    other_models: list[str] | None,
)

The AI models available for use by an OpenAI client, grouped by type.

Attributes:

Name	Type	Description
image_models	list[str]	Image generation or manipulation models.
other_models	list[str] | None	All other model types.
speech_models	list[str]	Text-to-speech models.
text_models	list[str]	Text completion models.

image_models instance-attribute

image_models: list[str]

Image generation or manipulation models.

other_models instance-attribute

other_models: list[str] | None

All other model types.

speech_models instance-attribute

speech_models: list[str]

Text-to-speech models.

text_models instance-attribute

text_models: list[str]

Text completion models.

execute_request_with_retry async

execute_request_with_retry(
    func: Callable[..., Awaitable[T]],
    *args: Any,
    max_retries: int = 6,
    initial_delay_ms: float = 1000,
    exponential_base: float = 2,
    jitter: bool = True,
    **kwargs: Any
) -> T

Calls an async function and retries with exponential backoff on RateLimitError.

This method parses the rate-limit-specific wait time from the OpenAI error and uses random exponential backoff to avoid hammering the API in tight loops.

Parameters:

Name	Type	Description	Default
func	Callable[..., Awaitable[T]]	The function to call.	required
*args	Any	Positional arguments to pass to the function.	()
max_retries	int	The maximum number of retries before giving up.	6
initial_delay_ms	float	The initial delay in milliseconds before the first retry.	1000
exponential_base	float	The exponential growth factor for delay intervals.	2
jitter	bool	Whether to apply random jitter to the delay interval.	True
**kwargs	Any	Keyword arguments to pass to the function.	{}

Returns:

Type	Description
T	The awaited result of func.

Raises:

Type	Description
Exception	If max_retries is exceeded.

get_usable_models_async async

get_usable_models_async() -> AIModels

Asynchronously get the usable models, fetching them if not already cached.

get_usable_models_sync

get_usable_models_sync() -> AIModels

Synchronously get the usable models using asyncio.run().