scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-15 23:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 165

(sap) add a possibility to put the service key by both variants

Browse source
This commit is contained in:
Vasilisa Parshikova 2026-03-11 16:06:33 +04:00 • committed by Sameer Kankute
parent a7d23d638d
commit 61a3bd1d25
No known key found for this signature in database
1 changed files with 48 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

78
litellm/llms/sap/credentials.py
View file

@ -190,11 +190,25 @@ def resolve_resource_group(sources: List[Source]) -> Optional[str]:
rg_cred = CredentialsValue("resource_group", default="default")
for source in sources:
value = source.get(rg_cred)
if value:
if value is not None:
verbose_logger.debug(f"Resolved GEN AI Hub resource_group from source {source.name}")
return value
return rg_cred.default
def _get_function_to_resolve_old_documented_and_correct_service_key(
service_key: Union[str, dict], cv: CredentialsValue):
if service_key is not None and "credentials" in service_key:
return _str_or_none(
_get_nested(service_key, (("credentials",) + cv.vcap_key) if cv.vcap_key else (cv.name,))
)
else:
return _str_or_none(
_get_nested(service_key, cv.vcap_key if cv.vcap_key else (cv.name,))
) if service_key else None
def fetch_credentials(service_key: Optional[Union[str, dict]] = None, profile: Optional[str] = None, **kwargs) -> Dict[str, str]:
"""
Resolution order (first-source-wins):
@ -220,12 +234,13 @@ def fetch_credentials(service_key: Optional[Union[str, dict]] = None, profile: O
Source("kwargs",
lambda cv: _str_or_none(kwargs.get(cv.name))),
Source("service key",
lambda cv: _str_or_none(_get_nested(service_key, cv.vcap_key if cv.vcap_key else (cv.name,)))
if service_key else None), # type: ignore[arg-type]
lambda cv: _get_function_to_resolve_old_documented_and_correct_service_key(service_key, cv)), # type: ignore[arg-type]
Source("environment variables",
lambda cv: _str_or_none(os.environ.get(f'AICORE_{cv.name.upper()}'))),
Source("config file",
lambda cv: _str_or_none(config.get(f'AICORE_{cv.name.upper()}') or config.get(cv.name))),
lambda cv: _str_or_none(config.get(f'AICORE_{cv.name.upper()}')
if config.get(f'AICORE_{cv.name.upper()}') is not None
else config.get(cv.name))),
Source("VCAP service",
lambda cv: _str_or_none(
_get_nested(vcap_service, (("credentials",) + cv.vcap_key) if cv.vcap_key else (cv.name,))
@ -273,6 +288,30 @@ def validate_credentials(
"(cert_str & key_str), or (cert_file_path & key_file_path)."
)
def _request_token(client_id:str, auth_url: str, timeout:float, cert_pair=None, client_secret=None) -> tuple[str, datetime]:
data = {"grant_type": "client_credentials", "client_id": client_id}
if client_secret:
data["client_secret"] = client_secret
resp: Optional[httpx.Response] = None
try:
if cert_pair:
with httpx.Client(cert=cert_pair) as raw_client:
handler = HTTPHandler(client=raw_client)
resp = handler.post(auth_url, data=data, timeout=timeout) # type: ignore[arg-type]
payload = resp.json()
else:
handler = _get_httpx_client()
resp = handler.post(auth_url, data=data, timeout=timeout) # type: ignore[arg-type]
payload = resp.json()
access_token = payload["access_token"]
expires_in = int(payload.get("expires_in", 3600))
expiry_date = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
return f"Bearer {access_token}", expiry_date
except Exception as e:
msg = resp.text if resp is not None else getattr(e, "text", str(e))
raise RuntimeError(f"Token request failed: {msg}") from e
def get_token_creator(
service_key: Optional[Union[str, dict]] = None,
profile: Optional[str] = None,
@ -319,33 +358,10 @@ def get_token_creator(
token: Optional[str] = None
token_expiry: Optional[datetime] = None
def _request_token(cert_pair=None) -> tuple[str, datetime]:
data = {"grant_type": "client_credentials", "client_id": client_id}
if client_secret:
data["client_secret"] = client_secret
resp: Optional[httpx.Response] = None
try:
if cert_pair:
with httpx.Client(cert=cert_pair) as raw_client:
handler = HTTPHandler(client=raw_client)
resp = handler.post(auth_url, data=data, timeout=timeout) # type: ignore[arg-type]
else:
handler = _get_httpx_client()
resp = handler.post(auth_url, data=data, timeout=timeout) # type: ignore[arg-type]
payload = resp.json()
access_token = payload["access_token"]
expires_in = int(payload.get("expires_in", 3600))
expiry_date = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
return f"Bearer {access_token}", expiry_date
except Exception as e:
msg = resp.text if resp is not None else getattr(e, "text", str(e))
raise RuntimeError(f"Token request failed: {msg}") from e
def _fetch_token() -> tuple[str, datetime]:
# Case 1: secret-based auth
if client_secret:
return _request_token()
return _request_token(auth_url=auth_url, client_id=client_id, timeout=timeout, client_secret=client_secret)
# Case 2: cert/key strings
if cert_str and key_str:
cert_str_fixed = cert_str.replace("\\n", "\n")
@ -357,9 +373,11 @@ def get_token_creator(
f.write(cert_str_fixed)
with open(key_path, "w") as f:
f.write(key_str_fixed)
return _request_token(cert_pair=(cert_path, key_path))
return _request_token(auth_url=auth_url, client_id=client_id, timeout=timeout,
cert_pair=(cert_path, key_path))
# Case 3: file-based cert/key
return _request_token(cert_pair=(cert_file_path, key_file_path))
return _request_token(auth_url=auth_url, client_id=client_id, timeout=timeout,
cert_pair=(cert_file_path, key_file_path))
def get_token() -> str:
nonlocal token, token_expiry