litellm-mirror/litellm/llms/fireworks_ai/common_utils.py
Krish Dholakia 6ba3c4a4f8
VertexAI non-jsonl file storage support (#9781)
* test: add initial e2e test

* fix(vertex_ai/files): initial commit adding sync file create support

* refactor: initial commit of vertex ai non-jsonl files reaching gcp endpoint

* fix(vertex_ai/files/transformation.py): initial working commit of non-jsonl file call reaching backend endpoint

* fix(vertex_ai/files/transformation.py): working e2e non-jsonl file upload

* test: working e2e jsonl call

* test: unit testing for jsonl file creation

* fix(vertex_ai/transformation.py): reset file pointer after read

allow multiple reads on same file object

* fix: fix linting errors

* fix: fix ruff linting errors

* fix: fix import

* fix: fix linting error

* fix: fix linting error

* fix(vertex_ai/files/transformation.py): fix linting error

* test: update test

* test: update tests

* fix: fix linting errors

* fix: fix test

* fix: fix linting error
2025-04-09 14:01:48 -07:00

52 lines
1.5 KiB
Python

from typing import List, Optional, Union
from httpx import Headers
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
from ..base_llm.chat.transformation import BaseLLMException
class FireworksAIException(BaseLLMException):
pass
class FireworksAIMixin:
"""
Common Base Config functions across Fireworks AI Endpoints
"""
def get_error_class(
self, error_message: str, status_code: int, headers: Union[dict, Headers]
) -> BaseLLMException:
return FireworksAIException(
status_code=status_code,
message=error_message,
headers=headers,
)
def _get_api_key(self, api_key: Optional[str]) -> Optional[str]:
dynamic_api_key = api_key or (
get_secret_str("FIREWORKS_API_KEY")
or get_secret_str("FIREWORKS_AI_API_KEY")
or get_secret_str("FIREWORKSAI_API_KEY")
or get_secret_str("FIREWORKS_AI_TOKEN")
)
return dynamic_api_key
def validate_environment(
self,
headers: dict,
model: str,
messages: List[AllMessageValues],
optional_params: dict,
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> dict:
api_key = self._get_api_key(api_key)
if api_key is None:
raise ValueError("FIREWORKS_API_KEY is not set")
return {"Authorization": "Bearer {}".format(api_key), **headers}