Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/artifact.py: 24%
71 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1from __future__ import annotations 1a
3from typing import Optional 1a
4from uuid import UUID 1a
6import orjson 1a
7import typer 1a
8from rich.pretty import Pretty 1a
9from rich.table import Table 1a
11from prefect.cli._types import PrefectTyper 1a
12from prefect.cli._utilities import exit_with_error, exit_with_success 1a
13from prefect.cli.root import app, is_interactive 1a
14from prefect.client.orchestration import get_client 1a
15from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey 1a
16from prefect.client.schemas.sorting import ArtifactCollectionSort, ArtifactSort 1a
17from prefect.exceptions import ObjectNotFound 1a
18from prefect.types._datetime import human_friendly_diff 1a
20artifact_app: PrefectTyper = PrefectTyper( 1a
21 name="artifact", help="Inspect and delete artifacts."
22)
23app.add_typer(artifact_app) 1a
26@artifact_app.command("ls") 1a
27async def list_artifacts( 1a
28 limit: int = typer.Option(
29 100,
30 "--limit",
31 help="The maximum number of artifacts to return.",
32 ),
33 all: bool = typer.Option(
34 False,
35 "--all",
36 "-a",
37 help="Whether or not to only return the latest version of each artifact.",
38 ),
39):
40 """
41 List artifacts.
42 """
43 table = Table(
44 title="Artifacts",
45 caption="List Artifacts using `prefect artifact ls`",
46 show_header=True,
47 )
49 table.add_column("ID", justify="right", style="cyan", no_wrap=True)
50 table.add_column("Key", style="blue", no_wrap=True)
51 table.add_column("Type", style="blue", no_wrap=True)
52 table.add_column("Updated", style="blue", no_wrap=True)
54 async with get_client() as client:
55 if all:
56 artifacts = await client.read_artifacts(
57 sort=ArtifactSort.KEY_ASC,
58 limit=limit,
59 )
61 for artifact in sorted(artifacts, key=lambda x: f"{x.key}"):
62 updated = (
63 human_friendly_diff(artifact.updated) if artifact.updated else ""
64 )
65 table.add_row(
66 str(artifact.id),
67 artifact.key,
68 artifact.type,
69 updated,
70 )
72 else:
73 artifacts = await client.read_latest_artifacts(
74 sort=ArtifactCollectionSort.KEY_ASC,
75 limit=limit,
76 )
78 for artifact in sorted(artifacts, key=lambda x: f"{x.key}"):
79 updated = (
80 human_friendly_diff(artifact.updated) if artifact.updated else ""
81 )
82 table.add_row(
83 str(artifact.latest_id),
84 artifact.key,
85 artifact.type,
86 updated,
87 )
89 app.console.print(table)
92@artifact_app.command("inspect") 1a
93async def inspect( 1a
94 key: str,
95 limit: int = typer.Option(
96 10,
97 "--limit",
98 help="The maximum number of artifacts to return.",
99 ),
100 output: Optional[str] = typer.Option(
101 None,
102 "--output",
103 "-o",
104 help="Specify an output format. Currently supports: json",
105 ),
106):
107 """
108 View details about an artifact.
110 Arguments:
111 key: the key of the artifact to inspect
113 Examples:
114 `$ prefect artifact inspect "my-artifact"`
115 ```json
116 [
117 {
118 'id': 'ba1d67be-0bd7-452e-8110-247fe5e6d8cc',
119 'created': '2023-03-21T21:40:09.895910+00:00',
120 'updated': '2023-03-21T21:40:09.895910+00:00',
121 'key': 'my-artifact',
122 'type': 'markdown',
123 'description': None,
124 'data': 'my markdown',
125 'metadata_': None,
126 'flow_run_id': '8dc54b6f-6e24-4586-a05c-e98c6490cb98',
127 'task_run_id': None
128 },
129 {
130 'id': '57f235b5-2576-45a5-bd93-c829c2900966',
131 'created': '2023-03-27T23:16:15.536434+00:00',
132 'updated': '2023-03-27T23:16:15.536434+00:00',
133 'key': 'my-artifact',
134 'type': 'markdown',
135 'description': 'my-artifact-description',
136 'data': 'my markdown',
137 'metadata_': None,
138 'flow_run_id': 'ffa91051-f249-48c1-ae0f-4754fcb7eb29',
139 'task_run_id': None
140 }
141 ]
142 ```
143 """
144 if output and output.lower() != "json":
145 exit_with_error("Only 'json' output format is supported.")
147 async with get_client() as client:
148 artifacts = await client.read_artifacts(
149 limit=limit,
150 sort=ArtifactSort.UPDATED_DESC,
151 artifact_filter=ArtifactFilter(key=ArtifactFilterKey(any_=[key])),
152 )
153 if not artifacts:
154 exit_with_error(f"Artifact {key!r} not found.")
156 artifacts_json = [a.model_dump(mode="json") for a in artifacts]
158 if output and output.lower() == "json":
159 json_output = orjson.dumps(
160 artifacts_json, option=orjson.OPT_INDENT_2
161 ).decode()
162 app.console.print(json_output)
163 else:
164 app.console.print(Pretty(artifacts_json))
167@artifact_app.command("delete") 1a
168async def delete( 1a
169 key: Optional[str] = typer.Argument(
170 None, help="The key of the artifact to delete."
171 ),
172 artifact_id: Optional[UUID] = typer.Option(
173 None, "--id", help="The ID of the artifact to delete."
174 ),
175):
176 """
177 Delete an artifact.
179 Arguments:
180 key: the key of the artifact to delete
182 Examples:
183 `$ prefect artifact delete "my-artifact"`
184 """
185 if key and artifact_id:
186 exit_with_error("Please provide either a key or an artifact_id but not both.")
188 async with get_client() as client:
189 if artifact_id is not None:
190 try:
191 if is_interactive() and not typer.confirm(
192 (
193 "Are you sure you want to delete artifact with id"
194 f" {str(artifact_id)!r}?"
195 ),
196 default=False,
197 ):
198 exit_with_error("Deletion aborted.")
200 await client.delete_artifact(artifact_id)
201 exit_with_success(f"Deleted artifact with id {str(artifact_id)!r}.")
202 except ObjectNotFound:
203 exit_with_error(f"Artifact with id {str(artifact_id)!r} not found!")
205 elif key is not None:
206 artifacts = await client.read_artifacts(
207 artifact_filter=ArtifactFilter(key=ArtifactFilterKey(any_=[key])),
208 )
209 if not artifacts:
210 exit_with_error(
211 f"Artifact with key {key!r} not found. You can also specify an"
212 " artifact id with the --id flag."
213 )
215 if is_interactive() and not typer.confirm(
216 (
217 f"Are you sure you want to delete {len(artifacts)} artifact(s) with"
218 f" key {key!r}?"
219 ),
220 default=False,
221 ):
222 exit_with_error("Deletion aborted.")
224 for a in artifacts:
225 await client.delete_artifact(a.id)
227 exit_with_success(f"Deleted {len(artifacts)} artifact(s) with key {key!r}.")
229 else:
230 exit_with_error("Please provide a key or an artifact_id.")