artifact_query.py
4.31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
"""c_artifact_get (`GET /api/artifacts/:artifact_id`, named in
docs/sdd/sdd-mvcp-contract.json but never documented in 03-C's prose, and
never implemented) + c_artifact_list (no operation_id in either the prose
doc or the MVCP JSON at all, but v_artifact_browser's own filter fields --
`artifact_type_filter`/`task_id_filter`, 02-V §2.8 -- require one; same
"unowned gap" precedent as c_task_list before U-025 built it). Returns a
bare array from list_artifacts(), matching list_tasks()'s shape rather than
the MVCP JSON's c_task_list `{items, total}` wrapper -- no pagination need
for a local single-project MVP, and staying consistent with this
codebase's own established list-endpoint shape matters more here than
matching a machine-generated mirror of the SDD.
"""
import json
import os
class ArtifactNotFoundError(Exception):
"""NOT_FOUND -- c_artifact_get on a nonexistent artifact_id."""
def _row_to_dict(r):
return {
"artifact_id": r[0], "artifact_type": r[1], "task_id": r[2], "produced_by_run_id": r[3],
"file_path": r[4], "git_commit_sha": r[5],
"depends_on_artifact_ids": json.loads(r[6]) if r[6] else [],
"validation_status": r[7], "created_at": r[8],
}
def list_artifacts(conn, project_id, task_id=None, artifact_type=None):
query = (
"SELECT a.artifact_id, a.artifact_type, a.task_id, a.produced_by_run_id, a.file_path, "
"a.git_commit_sha, a.depends_on_artifact_ids, a.validation_status, a.created_at "
"FROM artifact a JOIN task t ON t.task_id = a.task_id WHERE t.project_id = ?"
)
params = [project_id]
if task_id:
query += " AND a.task_id = ?"
params.append(task_id)
if artifact_type:
query += " AND a.artifact_type = ?"
params.append(artifact_type)
query += " ORDER BY a.created_at DESC"
rows = conn.execute(query, params).fetchall()
return [_row_to_dict(r) for r in rows]
def get_artifact(conn, artifact_id):
"""Returns the artifact row plus its file content (read live off disk --
`artifact` itself has no content column, see 01-M 1.5) and its lineage
chain, so v_artifact_browser's preview pane needs exactly one request."""
row = conn.execute(
"SELECT a.artifact_id, a.artifact_type, a.task_id, a.produced_by_run_id, a.file_path, "
"a.git_commit_sha, a.depends_on_artifact_ids, a.validation_status, a.created_at, p.target_folder_path "
"FROM artifact a JOIN task t ON t.task_id = a.task_id JOIN project p ON p.project_id = t.project_id "
"WHERE a.artifact_id = ?",
(artifact_id,),
).fetchone()
if row is None:
raise ArtifactNotFoundError(artifact_id)
result = _row_to_dict(row[:9])
abs_path = os.path.join(row[9], row[4])
try:
with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
result["content"] = f.read()
except OSError:
result["content"] = None
result["lineage"] = lineage_chain(conn, artifact_id)
return result
def lineage_chain(conn, artifact_id, max_depth=50):
"""02-V §2.8: click_row 展示 depends_on_artifact_ids 血缘链. Returns
oldest-first [{artifact_id, artifact_type}, ...] ending with
`artifact_id` itself. `depends_on_artifact_ids` is a list (an artifact
can depend on more than one predecessor), but every real usage in this
pipeline is a linear brd->prd->sdd chain and the wireframe itself only
ever shows a single line -- walks depends_on_artifact_ids[0] when more
than one dependency exists, a deliberate simplification for a case this
codebase doesn't otherwise produce. A cycle guard is defensive: this is
a read path and shouldn't assume c_artifact_write's own dependency
validation was never bypassed (e.g. by a direct DB edit)."""
chain = []
seen = set()
current_id = artifact_id
while current_id and current_id not in seen and len(chain) < max_depth:
seen.add(current_id)
row = conn.execute(
"SELECT artifact_id, artifact_type, depends_on_artifact_ids FROM artifact WHERE artifact_id = ?",
(current_id,),
).fetchone()
if row is None:
break
chain.append({"artifact_id": row[0], "artifact_type": row[1]})
deps = json.loads(row[2]) if row[2] else []
current_id = deps[0] if deps else None
chain.reverse()
return chain