o ưi!@s\dZddlmZddlmZmZmZmZddlZddl m Z ddl m Z Gddde Z dS) ag Custom Secret Manager Integration This module provides a base class for implementing custom secret managers in LiteLLM. Usage: from litellm.integrations.custom_secret_manager import CustomSecretManager class MySecretManager(CustomSecretManager): def __init__(self): super().__init__(secret_manager_name="my_secret_manager") async def async_read_secret( self, secret_name: str, optional_params=None, timeout=None, ): # Your implementation here return await self._fetch_secret_from_service(secret_name) def sync_read_secret( self, secret_name: str, optional_params=None, timeout=None, ): # Your implementation here return self._fetch_secret_from_service_sync(secret_name) # Set your custom secret manager import litellm from litellm.types.secret_managers.main import KeyManagementSystem litellm.secret_manager_client = MySecretManager() litellm._key_management_system = KeyManagementSystem.CUSTOM )abstractmethod)AnyDictOptionalUnionN)verbose_logger)BaseSecretManagercsxeZdZdZ ddeeffdd Ze  ddedeedee e e j fd eefd d Z e  ddedeedee e e j fd eefd d Z    ddededeedeedee e e j fdee eefd eeeffddZ   d dedeedeedee e e j fd ef ddZd efddZ ddee e e j fd efddZd efddZZS)!CustomSecretManageray Base class for implementing custom secret managers. This class provides a standard interface for implementing custom secret management integrations in LiteLLM. Users can extend this class to integrate their own secret management systems. Example: ```python from litellm.integrations.custom_secret_manager import CustomSecretManager class MyVaultSecretManager(CustomSecretManager): def __init__(self, vault_url: str, token: str): super().__init__(secret_manager_name="my_vault") self.vault_url = vault_url self.token = token async def async_read_secret(self, secret_name: str, optional_params=None, timeout=None): # Implementation for reading secrets from your vault async with httpx.AsyncClient() as client: response = await client.get( f"{self.vault_url}/v1/secret/{secret_name}", headers={"X-Vault-Token": self.token}, timeout=timeout ) return response.json()["data"]["value"] def sync_read_secret(self, secret_name: str, optional_params=None, timeout=None): # Sync implementation with httpx.Client() as client: response = client.get( f"{self.vault_url}/v1/secret/{secret_name}", headers={"X-Vault-Token": self.token}, timeout=timeout ) return response.json()["data"]["value"] ``` Nsecret_manager_namec s"t|pd|_tddS)a5 Initialize the CustomSecretManager. Args: secret_manager_name: A descriptive name for your secret manager. This is used for logging and debugging purposes. **kwargs: Additional keyword arguments to pass to your secret manager. Zcustom_secret_managerz!Initialized custom secret managerN)super__init__r rinfo)selfr kwargs __class__a/home/app/Keep/.python/lib/python3.10/site-packages/litellm/integrations/custom_secret_manager.pyr Xs zCustomSecretManager.__init__ secret_nameoptional_paramstimeoutreturncsdS)a Asynchronously read a secret from your custom secret manager. Args: secret_name: Name/path of the secret to read optional_params: Additional parameters specific to your secret manager timeout: Request timeout Returns: The secret value if found, None otherwise Raises: Exception: If there's an error reading the secret Nrrrrrrrrasync_read_secretksz%CustomSecretManager.async_read_secretcCsdS)a Synchronously read a secret from your custom secret manager. Args: secret_name: Name/path of the secret to read optional_params: Additional parameters specific to your secret manager timeout: Request timeout Returns: The secret value if found, None otherwise Raises: Exception: If there's an error reading the secret Nrrrrrsync_read_secretsz$CustomSecretManager.sync_read_secret secret_value descriptiontagsctd|jd)a Asynchronously write a secret to your custom secret manager. This is optional to implement. If your secret manager supports writing secrets, you can override this method. Args: secret_name: Name/path of the secret to write secret_value: Value to store description: Description of the secret optional_params: Additional parameters specific to your secret manager timeout: Request timeout tags: Optional tags to apply to the secret Returns: Response from the secret manager containing write operation details Raises: NotImplementedError: If write operations are not supported z)Write operations are not implemented for z5. Override async_write_secret() to add write support.NotImplementedErrorr )rrrrrrrrrrasync_write_secrets z&CustomSecretManager.async_write_secretrecovery_window_in_dayscr)a Asynchronously delete a secret from your custom secret manager. This is optional to implement. If your secret manager supports deleting secrets, you can override this method. Args: secret_name: Name of the secret to delete recovery_window_in_days: Number of days before permanent deletion (if supported) optional_params: Additional parameters specific to your secret manager timeout: Request timeout Returns: Response from the secret manager containing deletion details Raises: NotImplementedError: If delete operations are not supported z*Delete operations are not implemented for z7. Override async_delete_secret() to add delete support.r)rrr#rrrrrasync_delete_secrets z'CustomSecretManager.async_delete_secretcCstddS)a= Validate that all required environment variables and configuration are present. Override this method to validate your secret manager's configuration. Returns: True if the environment is valid Raises: ValueError: If required configuration is missing z>No environment validation configured for custom secret managerT)rdebugrrrrvalidate_environments z(CustomSecretManager.validate_environmentcstd|jdS)a# Perform a health check on your secret manager. This is optional to implement. Override this method to add health check support. Args: timeout: Request timeout Returns: True if the secret manager is healthy, False otherwise z!Health check not implemented for T)rr%r )rrrrrasync_health_checks  z&CustomSecretManager.async_health_checkcCsd|jjd|jdS)N)r__name__r r&rrr__repr__szCustomSecretManager.__repr__)N)NN)NNNN)r"NN)r* __module__ __qualname____doc__rstrr rdictrfloathttpxTimeoutrrlistrrr!intr$boolr'r(r+ __classcell__rrrrr 0s)    %  r )r.abcrtypingrrrrr2Zlitellm._loggingrZ+litellm.secret_managers.base_secret_managerrr rrrrs &