Source code for ibm_watsonx_ai.gateway.providers

#  -----------------------------------------------------------------------------------------
#  (C) Copyright IBM Corp. 2025.
#  https://opensource.org/licenses/BSD-3-Clause
#  -----------------------------------------------------------------------------------------

import pandas as pd

from ibm_watsonx_ai import APIClient
from ibm_watsonx_ai.wml_client_error import WMLClientError
from ibm_watsonx_ai.wml_resource import WMLResource


[docs] class Providers(WMLResource): """Model Gateway providers class.""" def __init__(self, api_client: APIClient): WMLResource.__init__(self, __name__, api_client)
[docs] def create( self, provider: str, name: str | None = None, data: dict | None = None ) -> dict: """Create provider in Model Gateway. :param provider: provider name :type provider: str :param name: name of provider for display :type name: str, optional :param data: data required to connect to provider api :type data: dict, optional :returns: provider details :rtype: dict """ request_json = {} if name: request_json["name"] = name if data is not None: request_json["data"] = data response = self._client.httpx_client.post( self._client._href_definitions.get_gateway_provider_href(provider), headers=self._client._get_headers(), json=request_json, ) return self._handle_response(201, "provider creation", response)
[docs] def get_details(self, provider_id: str | None = None) -> dict: """Get provider/providers details: - `provider_id` is set - details for given provider are returned - `provider_id` is `None` - details for all providers are returned :param provider_id: unique provider ID :type provider_id: str, optional :returns: provider/providers details :rtype: dict """ if provider_id: response = self._client.httpx_client.get( self._client._href_definitions.get_gateway_provider_href(provider_id), headers=self._client._get_headers(), ) return self._handle_response(200, "getting provider details", response) else: response = self._client.httpx_client.get( self._client._href_definitions.get_gateway_providers_href(), headers=self._client._get_headers(), ) return self._handle_response(200, "getting providers details", response)
[docs] def get_available_models_details(self, provider_id: str) -> dict: """Get available models details for given provider. :param provider_id: unique provider ID :type provider_id: str :returns: details of available models for provider :rtype: dict """ response = self._client.httpx_client.get( self._client._href_definitions.get_gateway_provider_available_models_href( provider_id ), headers=self._client._get_headers(), ) return self._handle_response( 200, "getting provider available models details", response )
[docs] def list(self) -> pd.DataFrame: """List providers. :returns: dataframe with providers details :rtype: pandas.DataFrame """ providers_details = self.get_details()["data"] providers_values = [ ( m["uuid"], m["name"], m["type"], ) for m in providers_details ] table = self._list(providers_values, ["ID", "NAME", "TYPE"], limit=None) return table
[docs] def list_available_models(self, provider_id: str) -> pd.DataFrame: """List available models for provider. :param provider_id: unique provider ID :type provider_id: str :returns: dataframe with available models details :rtype: pandas.DataFrame """ models_details = self.get_available_models_details(provider_id)["data"] if models_details is None: raise WMLClientError( f"Available models not supported for provider=`{provider_id}`." ) models_values = [ ( m["id"], m["owned_by"], ) for m in models_details ] table = self._list(models_values, ["MODEL_ID", "TYPE"], limit=None) return table
[docs] def delete(self, provider_id: str) -> str: """Delete provider. :param provider_id: unique provider ID :type provider_id: str :return: status ("SUCCESS" if succeeded) :rtype: str """ response = self._client.httpx_client.delete( self._client._href_definitions.get_gateway_provider_href(provider_id), headers=self._client._get_headers(), ) return self._handle_response( 204, "provider deletion", response, json_response=False )
[docs] @staticmethod def get_id(provider_details: dict) -> str: """Get provider ID from provider details. :param provider_details: details of the provider in Model Gateway :type provider_details: dict :returns: unique provider ID :rtype: str """ return provider_details["uuid"]