o ưi?@sPddlmZmZddlmZmZmZmZddlZddl m Z GdddeZ dS))ABCabstractmethod)AnyDictOptionalUnionN)verbose_loggerc@s\eZdZdZe  ddedeedeee e j fdeefddZ e  ddedeedeee e j fdeefd d Z e    dded ed eedeedeee e j fd eeeefdeeeffddZe   ddedeedeedeee e j fdef ddZ  ddedededeedeee e j fdef ddZdS)BaseSecretManagerzD Abstract base class for secret management implementations. N secret_nameoptional_paramstimeoutreturncdS)a Asynchronously read a secret from the secret manager. Args: secret_name (str): Name/path of the secret to read optional_params (Optional[dict]): Additional parameters specific to the secret manager timeout (Optional[Union[float, httpx.Timeout]]): Request timeout Returns: Optional[str]: The secret value if found, None otherwise Nselfr r r rrb/home/app/Keep/.python/lib/python3.10/site-packages/litellm/secret_managers/base_secret_manager.pyasync_read_secretsz#BaseSecretManager.async_read_secretcCsdS)a Synchronously read a secret from the secret manager. Args: secret_name (str): Name/path of the secret to read optional_params (Optional[dict]): Additional parameters specific to the secret manager timeout (Optional[Union[float, httpx.Timeout]]): Request timeout Returns: Optional[str]: The secret value if found, None otherwise Nrrrrrsync_read_secret"sz"BaseSecretManager.sync_read_secret secret_value descriptiontagscr)a+ Asynchronously write a secret to the secret manager. Args: secret_name (str): Name/path of the secret to write secret_value (str): Value to store description (Optional[str]): Description of the secret. Some secret managers allow storing a description with the secret. optional_params (Optional[dict]): Additional parameters specific to the secret manager timeout (Optional[Union[float, httpx.Timeout]]): Request timeout tags: Optional dict or list of tags to apply, e.g. {"Environment": "Prod", "Owner": "AI-Platform"} or [{"Key": "Environment", "Value": "Prod"}] Returns: Dict[str, Any]: Response from the secret manager containing write operation details Nr)rr rrr r rrrrasync_write_secret6sz$BaseSecretManager.async_write_secretrecovery_window_in_dayscr)a Async function to delete a secret from the secret manager Args: secret_name: Name of the secret to delete recovery_window_in_days: Number of days before permanent deletion (default: 7) optional_params: Additional parameters specific to the secret manager timeout: Request timeout Returns: dict: Response from the secret manager containing deletion details Nr)rr rr r rrrasync_delete_secretQsz%BaseSecretManager.async_delete_secretcurrent_secret_namenew_secret_namenew_secret_valuec szL|j|||dIdH}|durtd|d|j||d|||dIdH}|j|||dIdH}|dur?td||j|d||d IdH|WStjym} ztd t| j j td | j j d} ~ wtj yxtd t y} z td t| d} ~ ww) ak Async function to rotate a secret by creating a new one and deleting the old one. This allows for both value and name changes during rotation. Args: current_secret_name: Current name of the secret new_secret_name: New name for the secret new_secret_value: New value for the secret optional_params: Additional AWS parameters timeout: Request timeout Returns: dict: Response containing the new secret details Raises: ValueError: If the secret doesn't exist or if there's an HTTP error )r r r NzCurrent secret z not foundz Rotated from )r rrr r zFailed to verify new secret r)r rr r z0Error rotating secret in AWS Secrets Manager: %szHTTP error occurred: zTimeout error occurred) r ValueErrorrrhttpxZHTTPStatusErrorr exceptionstrresponsetextZTimeoutException Exception) rrrrr r Z old_secretZcreate_responseZ new_secreterrerrrasync_rotate_secretgs\      z%BaseSecretManager.async_rotate_secret)NN)NNNN)rNN)__name__ __module__ __qualname____doc__rr"rdictrfloatr Timeoutrrlistrrrintrr(rrrrr s     r ) abcrrtypingrrrrr Zlitellmrr rrrrs