Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/variable.py: 29%
77 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1import json 1a
2from typing import Any, Dict, List, Optional, Union 1a
4import orjson 1a
5import typer 1a
6from rich.pretty import Pretty 1a
7from rich.table import Table 1a
8from typing_extensions import Annotated 1a
10from prefect.cli._types import PrefectTyper 1a
11from prefect.cli._utilities import exit_with_error, exit_with_success 1a
12from prefect.cli.root import app, is_interactive 1a
13from prefect.client.orchestration import get_client 1a
14from prefect.client.schemas.actions import VariableCreate, VariableUpdate 1a
15from prefect.exceptions import ObjectNotFound 1a
16from prefect.types._datetime import human_friendly_diff 1a
18variable_app: PrefectTyper = PrefectTyper(name="variable", help="Manage variables.") 1a
19app.add_typer(variable_app) 1a
22@variable_app.command("ls") 1a
23async def list_variables( 1a
24 limit: int = typer.Option(
25 100,
26 "--limit",
27 help="The maximum number of variables to return.",
28 ),
29):
30 """
31 List variables.
32 """
33 async with get_client() as client:
34 variables = await client.read_variables(
35 limit=limit,
36 )
38 table = Table(
39 title="Variables",
40 caption="List Variables using `prefect variable ls`",
41 show_header=True,
42 )
44 table.add_column("Name", style="blue", no_wrap=True)
45 # values can be up 5000 characters so truncate early
46 table.add_column("Value", style="blue", no_wrap=True, max_width=50)
47 table.add_column("Created", style="blue", no_wrap=True)
48 table.add_column("Updated", style="blue", no_wrap=True)
50 for variable in sorted(variables, key=lambda x: f"{x.name}"):
51 assert variable.created is not None, "created is not None"
52 assert variable.updated is not None, "updated is not None"
53 table.add_row(
54 variable.name,
55 json.dumps(variable.value),
56 human_friendly_diff(variable.created),
57 human_friendly_diff(variable.updated),
58 )
60 app.console.print(table)
63@variable_app.command("inspect") 1a
64async def inspect( 1a
65 name: str,
66 output: Optional[str] = typer.Option(
67 None,
68 "--output",
69 "-o",
70 help="Specify an output format. Currently supports: json",
71 ),
72):
73 """
74 View details about a variable.
76 Arguments:
77 name: the name of the variable to inspect
78 """
79 if output and output.lower() != "json":
80 exit_with_error("Only 'json' output format is supported.")
82 async with get_client() as client:
83 variable = await client.read_variable_by_name(
84 name=name,
85 )
86 if not variable:
87 exit_with_error(f"Variable {name!r} not found.")
89 if output and output.lower() == "json":
90 variable_json = variable.model_dump(mode="json")
91 json_output = orjson.dumps(
92 variable_json, option=orjson.OPT_INDENT_2
93 ).decode()
94 app.console.print(json_output)
95 else:
96 app.console.print(Pretty(variable))
99@variable_app.command("get") 1a
100async def get( 1a
101 name: str,
102):
103 """
104 Get a variable's value.
106 Arguments:
107 name: the name of the variable to get
108 """
110 async with get_client() as client:
111 variable = await client.read_variable_by_name(
112 name=name,
113 )
114 if variable:
115 app.console.print(json.dumps(variable.value))
116 else:
117 exit_with_error(f"Variable {name!r} not found.")
120def parse_value( 1a
121 value: str,
122) -> Union[str, int, float, bool, None, Dict[str, Any], List[str]]:
123 try:
124 parsed_value = json.loads(value)
125 except json.JSONDecodeError:
126 parsed_value = value
127 return parsed_value
130@variable_app.command("set") 1a
131async def _set( 1a
132 name: str,
133 value: str,
134 overwrite: bool = typer.Option(
135 False,
136 "--overwrite",
137 help="Overwrite the variable if it already exists.",
138 ),
139 tag: Annotated[
140 Optional[List[str]], typer.Option(help="Tag to associate with the variable.")
141 ] = None,
142):
143 """
144 Set a variable.
146 If the variable already exists, use `--overwrite` to update it.
148 Arguments:
149 name: the name of the variable to set
150 value: the value of the variable to set
151 --overwrite: overwrite the variable if it already exists
152 --tag: tag to associate with the variable (you may pass multiple)
153 """
155 async with get_client() as client:
156 variable = await client.read_variable_by_name(name)
157 var_dict = {"name": name, "value": parse_value(value), "tags": tag or []}
158 if variable:
159 if not overwrite:
160 exit_with_error(
161 f"Variable {name!r} already exists. Use `--overwrite` to update it."
162 )
163 await client.update_variable(VariableUpdate(**var_dict))
164 else:
165 await client.create_variable(VariableCreate(**var_dict))
167 exit_with_success(f"Set variable {name!r}.")
170@variable_app.command("unset", aliases=["delete"]) 1a
171async def unset( 1a
172 name: str,
173):
174 """
175 Unset a variable.
177 Arguments:
178 name: the name of the variable to unset
179 """
181 async with get_client() as client:
182 try:
183 if is_interactive() and not typer.confirm(
184 f"Are you sure you want to unset variable {name!r}?"
185 ):
186 exit_with_error("Unset aborted.")
187 await client.delete_variable_by_name(
188 name=name,
189 )
190 except ObjectNotFound:
191 exit_with_error(f"Variable {name!r} not found.")
193 exit_with_success(f"Unset variable {name!r}.")