project_query.py
3.46 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
"""c_project_list / c_project_get / c_project_update (2026-08-03, v_project_list):
02-V originally declared "不做完整多项目管理 UI（项目列表/切换）" as an MVP
non-goal -- this is a deliberate, explicit scope expansion beyond that,
requested directly by the user rather than found while implementing
something else. Same read/update-query pattern as task_query.py.
"Delete" maps to `project.status = 'archived'`, never a physical row
delete -- 01-M's own status enum (`active`/`archived`) already models this
as a soft-delete concept, and `project_id` is referenced via FK from
task/artifact/bug_record/work_graph_run/etc.; a literal DELETE would either
violate those constraints or need a destructive cascade that throws away
real audit history this codebase otherwise goes out of its way to
preserve (Git-tracked artifacts, bug records, ...). `c_project_create`
(project_scaffold.py) already handles "point at a folder that already has
real code in it" correctly -- it only ever creates `.graphlemon/` inside
the target folder and never touches anything else there -- so importing an
existing codebase needs no changes here, just this module's read/update
side.
"""
_VALID_STATUSES = ("active", "archived")
class ProjectNotFoundError(Exception):
"""NOT_FOUND."""
class InvalidParamError(Exception):
"""INVALID_PARAM."""
def list_projects(conn):
rows = conn.execute(
"SELECT project_id, name, target_folder_path, has_git_repo, status, created_at "
"FROM project ORDER BY created_at DESC"
).fetchall()
projects = []
for r in rows:
goal_count = conn.execute("SELECT COUNT(*) FROM task WHERE project_id = ?", (r[0],)).fetchone()[0]
projects.append({
"project_id": r[0], "name": r[1], "target_folder_path": r[2],
"has_git_repo": bool(r[3]), "status": r[4], "created_at": r[5], "goal_count": goal_count,
})
return projects
def get_project(conn, project_id):
row = conn.execute(
"SELECT project_id, name, target_folder_path, has_git_repo, graphlemon_subdir_initialized, "
"status, created_at FROM project WHERE project_id = ?",
(project_id,),
).fetchone()
if row is None:
raise ProjectNotFoundError(project_id)
goal_count = conn.execute("SELECT COUNT(*) FROM task WHERE project_id = ?", (project_id,)).fetchone()[0]
return {
"project_id": row[0], "name": row[1], "target_folder_path": row[2],
"has_git_repo": bool(row[3]), "graphlemon_subdir_initialized": bool(row[4]),
"status": row[5], "created_at": row[6], "goal_count": goal_count,
}
def update_project(conn, project_id, name=None, status=None):
"""Returns the updated project (same shape as get_project). Either
field may be omitted; a call with neither is a harmless no-op read."""
if conn.execute("SELECT 1 FROM project WHERE project_id = ?", (project_id,)).fetchone() is None:
raise ProjectNotFoundError(project_id)
if status is not None and status not in _VALID_STATUSES:
raise InvalidParamError("status must be one of {}, got {!r}".format(_VALID_STATUSES, status))
if name is not None:
if not name.strip():
raise InvalidParamError("name cannot be blank")
conn.execute("UPDATE project SET name = ? WHERE project_id = ?", (name, project_id))
if status is not None:
conn.execute("UPDATE project SET status = ? WHERE project_id = ?", (status, project_id))
conn.commit()
return get_project(conn, project_id)