Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/cli/commands/credentials.py: 0%

1import json 

2from collections.abc import Sequence 

3from typing import Final, Literal 

4 

5import click 

6import requests 

7import rich 

8from rich.table import Table 

9from typing_extensions import NotRequired, ReadOnly, TypedDict 

10 

11from ...credentials import CredentialsManagementClient 

12from ._cli_context import cli_context_values 

13 

14 

15class _CredentialInfo(TypedDict): 

16 custom_llm_provider: ReadOnly[NotRequired[str]] 

17 

18 

19class _CredentialItem(TypedDict): 

20 credential_name: ReadOnly[NotRequired[str]] 

21 credential_info: ReadOnly[NotRequired[_CredentialInfo]] 

22 

23 

24class _CredentialsListView(TypedDict): 

25 credentials: ReadOnly[Sequence[_CredentialItem]] 

26 

27 

28class _JsonObjectView(TypedDict): 

29 value: ReadOnly[dict[str, object]] 

30 

31 

32class _JsonBodyView(TypedDict): 

33 body: ReadOnly[object] 

34 

35 

36@click.group() 

37def credentials(): 

38 """Manage credentials for the LiteLLM proxy server""" 

39 

40 

41@credentials.command() 

42@click.option( 

43 "--format", 

44 "output_format", 

45 type=click.Choice(["table", "json"]), 

46 default="table", 

47 help="Output format (table or json)", 

48) 

49@click.pass_context 

50def list(ctx: click.Context, output_format: Literal["table", "json"]): 

51 """List all credentials""" 

52 context: Final = cli_context_values(ctx) 

53 client: Final = CredentialsManagementClient(context["base_url"], context["api_key"]) 

54 response: Final = client.list() 

55 assert isinstance(response, dict) 

56 

57 if output_format == "json": 

58 rich.print_json(data=response) 

59 else: # table format 

60 table: Final = Table(title="Credentials") 

61 

62 # Add columns 

63 table.add_column("Credential Name", style="cyan") 

64 table.add_column("Custom LLM Provider", style="green") 

65 

66 # Add rows 

67 listed: Final[_CredentialsListView] = {"credentials": response.get("credentials", [])} 

68 for cred in listed["credentials"]: 

69 info = cred.get("credential_info", {}) 

70 table.add_row( 

71 str(cred.get("credential_name", "")), 

72 str(info.get("custom_llm_provider", "")), 

73 ) 

74 

75 rich.print(table) 

76 

77 

78@credentials.command() 

79@click.argument("credential_name") 

80@click.option( 

81 "--info", 

82 type=str, 

83 help="JSON string containing credential info", 

84 required=True, 

85) 

86@click.option( 

87 "--values", 

88 type=str, 

89 help="JSON string containing credential values", 

90 required=True, 

91) 

92@click.pass_context 

93def create(ctx: click.Context, credential_name: str, info: str, values: str): 

94 """Create a new credential""" 

95 context: Final = cli_context_values(ctx) 

96 client: Final = CredentialsManagementClient(context["base_url"], context["api_key"]) 

97 try: 

98 credential_info: Final[_JsonObjectView] = {"value": json.loads(info)} 

99 credential_values: Final[_JsonObjectView] = {"value": json.loads(values)} 

100 except json.JSONDecodeError as e: 

101 raise click.BadParameter(f"Invalid JSON: {e}") 

102 

103 try: 

104 response: Final = client.create(credential_name, credential_info["value"], credential_values["value"]) 

105 rich.print_json(data=response) 

106 except requests.exceptions.HTTPError as e: 

107 click.echo(f"Error: HTTP {e.response.status_code}", err=True) 

108 try: 

109 error_body: Final[_JsonBodyView] = {"body": e.response.json()} 

110 rich.print_json(data=error_body["body"]) 

111 except json.JSONDecodeError: 

112 click.echo(e.response.text, err=True) 

113 raise click.Abort() 

114 

115 

116@credentials.command() 

117@click.argument("credential_name") 

118@click.pass_context 

119def delete(ctx: click.Context, credential_name: str): 

120 """Delete a credential by name""" 

121 context: Final = cli_context_values(ctx) 

122 client: Final = CredentialsManagementClient(context["base_url"], context["api_key"]) 

123 try: 

124 response: Final = client.delete(credential_name) 

125 rich.print_json(data=response) 

126 except requests.exceptions.HTTPError as e: 

127 click.echo(f"Error: HTTP {e.response.status_code}", err=True) 

128 try: 

129 error_body: Final[_JsonBodyView] = {"body": e.response.json()} 

130 rich.print_json(data=error_body["body"]) 

131 except json.JSONDecodeError: 

132 click.echo(e.response.text, err=True) 

133 raise click.Abort() 

134 

135 

136@credentials.command() 

137@click.argument("credential_name") 

138@click.pass_context 

139def get(ctx: click.Context, credential_name: str): 

140 """Get a credential by name""" 

141 context: Final = cli_context_values(ctx) 

142 client: Final = CredentialsManagementClient(context["base_url"], context["api_key"]) 

143 response: Final = client.get(credential_name) 

144 rich.print_json(data=response)