For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: 讓所有 GRC API 支援「一個專案 = 多個 AP」,API URL 加入 ap_uid 參數。
Architecture: 所有查詢 AP 控制項結構的 API URL 加入 /ap/<ap_uid> 路徑段。Repo 層查詢入口從 project_uid → ProjectAssessmentPlanMapping.first() → ap_id 改為 ap_uid → OscalAssessmentPlan → ap_id。Dashboard 改為取每個 project 最新 AP。新增 AP 列表 API 供前端選擇稽核輪次。
Tech Stack: Flask-RESTful, SQLAlchemy, dependency-injector, marshmallow, jedi-oscal
| 檔案 | 職責 |
|---|---|
app/grc/dto/assessment_plan_dto.py |
AP Menu DTO |
api/grc/serializers/assessment_plan.py |
AP Menu Response Schema |
api/grc/routes/assessment_plan_route.py |
AP Menu Route |
| 檔案 | 改動摘要 |
|---|---|
common/code/grc_error_code.py |
新增 GRC_AP_NOT_FOUND |
infra/grc/repository/grc_control_group_repo_impl.py |
查詢入口改用 ap_uid |
infra/grc/repository/grc_control_repo_impl.py |
查詢入口改用 ap_uid |
infra/grc/repository/grc_assessment_object_repo_impl.py |
查詢入口改用 ap_uid |
infra/grc/repository/grc_task_setup_repo_impl.py |
查詢入口改用 ap_uid |
infra/grc/repository/grc_review_repo_impl.py |
_resolve_control_ids() 改用 ap_uid |
infra/grc/repository/grc_dashboard_repo_impl.py |
取每個 project 最新 AP |
infra/grc/repository/grc_project_repo_impl.py |
新增 get_assessment_plans_menu() |
domain/grc/service/grc_control_group_domain_service.py |
參數 project_uid → ap_uid |
domain/grc/service/grc_control_domain_service.py |
參數 project_uid → ap_uid |
domain/grc/service/grc_assessment_object_domain_service.py |
參數 project_uid → ap_uid |
domain/grc/service/grc_task_setup_domain_service.py |
參數 project_uid → ap_uid |
domain/grc/service/grc_review_domain_service.py |
參數 project_uid → ap_uid |
domain/grc/service/grc_project_domain_service.py |
新增 get_assessment_plans_menu() |
app/grc/service/control_group_service.py |
參數 project_uid/project_id → ap_uid |
app/grc/service/control_service.py |
參數 project_uid → ap_uid |
app/grc/service/assessment_object_service.py |
參數 project_uid → ap_uid |
app/grc/service/task_setup_service.py |
參數 project_uid → ap_uid |
app/grc/service/review_service.py |
參數 project_uid → ap_uid |
app/grc/service/project_service.py |
新增 get_assessment_plans_menu() |
api/grc/routes/control_group_route.py |
URL + handler 加 ap_uid |
api/grc/routes/control_route.py |
URL + handler 加 ap_uid |
api/grc/routes/assessment_object_route.py |
URL + handler 加 ap_uid |
api/grc/routes/job_route.py |
URL + handler 加 ap_uid(list + create) |
api/grc/routes/task_setup_route.py |
URL + handler 加 ap_uid |
api/grc/routes/review_route.py |
URL + handler 加 ap_uid |
api/grc/__init__.py |
更新 URL + 註冊 AP route |
Files:
common/code/grc_error_code.pyapp/grc/dto/assessment_plan_dto.pyapi/grc/serializers/assessment_plan.py在 common/code/grc_error_code.py 的最後一筆 404 error 之後新增:
GRC_AP_NOT_FOUND = ("稽核計畫不存在", "GRC_404015")建立 app/grc/dto/assessment_plan_dto.py:
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class GrcAssessmentPlanMenuDto:
uid: str
name: str
round_number: int
status: Optional[str] = None
created_at: Optional[datetime] = None
@staticmethod
def from_row(row, round_number: int) -> "GrcAssessmentPlanMenuDto":
return GrcAssessmentPlanMenuDto(
uid=str(row.uid),
name=row.title or "",
round_number=round_number,
status=row.status,
created_at=row.created_at,
)建立 api/grc/serializers/assessment_plan.py:
from marshmallow import Schema, fields
class AssessmentPlanMenuResponseSchema(Schema):
uid = fields.String()
name = fields.String()
round_number = fields.Integer()
status = fields.String(allow_none=True)
created_at = fields.DateTime(allow_none=True)ap_uidFiles:
infra/grc/repository/grc_control_group_repo_impl.pyinfra/grc/repository/grc_control_repo_impl.pyinfra/grc/repository/grc_assessment_object_repo_impl.pyinfra/grc/repository/grc_task_setup_repo_impl.py所有 4 個 repo 的改動模式一致:把 project_uid → ProjectAssessmentPlanMapping.first() 替換為 ap_uid → OscalAssessmentPlan。
每個 repo 需要:
from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlanProjectAssessmentPlanMapping + Project.uid == project_uid 解析 ap_id 的查詢區塊ap_uid 查 OscalAssessmentPlan舊模式:
ap_row = (
session.query(
ProjectAssessmentPlanMapping.assessment_plan_id,
ProjectAssessmentPlanMapping.project_id,
)
.join(Project, Project.id == ProjectAssessmentPlanMapping.project_id)
.filter(Project.uid == project_uid)
.first()
)
if ap_row is None:
return None
ap_id, project_id = ap_row新模式:
ap = (
session.query(OscalAssessmentPlan)
.filter(OscalAssessmentPlan.uid == ap_uid)
.first()
)
if ap is None:
return None
ap_id = ap.id注意: 如果原本有解構出
project_id,需要確認後續是否用到。大多數情況project_id只用於 participant 查詢(task_setup_repo 需要),可以從ProjectAssessmentPlanMapping反查。
from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlanget_control_group_by_uid(project_uid, group_uid) → 改簽名為 get_control_group_by_uid(ap_uid, group_uid)
project_id(participant 查詢),改從 ProjectAssessmentPlanMapping 反查:
mapping = session.query(ProjectAssessmentPlanMapping).filter(
ProjectAssessmentPlanMapping.assessment_plan_id == ap_id
).first()
project_id = mapping.project_id if mapping else Nonelist_control_groups(project_uid, pager, sorts, <strong>filters) → 改簽名為 list_control_groups(ap_uid, pager, sorts, </strong>filters)
project_idfrom jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlanget_control_by_uid(project_uid, group_uid, control_uid) → get_control_by_uid(ap_uid, group_uid, control_uid)
list_controls(project_uid, group_uid, pager, sorts, **filters) → list_controls(ap_uid, group_uid, pager, sorts, **filters)
get_control_with_ao_detail(project_uid, group_uid, control_uid) → get_control_with_ao_detail(ap_uid, group_uid, control_uid)
ap 物件的 ssp_id 屬性可用from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlanlist_assessment_objects(project_uid, control_uid, pager, sorts, **filters) → list_assessment_objects(ap_uid, control_uid, pager, sorts, **filters)
ap_id,不需要 project_idfrom jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlanget_task_setup_tree(project_uid) → get_task_setup_tree(ap_uid)
project_id(participant 查詢)和 project_name(回傳 entity)ap = session.query(OscalAssessmentPlan).filter(OscalAssessmentPlan.uid == ap_uid).first()
if ap is None:
return GrcTaskSetupTreeEntity(project_uid="", project_name="", groups=[])
ap_id = ap.id
mapping = session.query(
ProjectAssessmentPlanMapping.project_id,
Project.uid.label("project_uid"),
Project.name.label("project_name"),
).join(Project, Project.id == ProjectAssessmentPlanMapping.project_id) \
.filter(ProjectAssessmentPlanMapping.assessment_plan_id == ap_id) \
.first()
project_id = mapping.project_id
project_uid = str(mapping.project_uid)
project_name = mapping.project_nameFiles:
infra/grc/repository/grc_review_repo_impl.pyinfra/grc/repository/grc_dashboard_repo_impl.pyReview repo 的 AP 解析在 helper method _resolve_control_ids() 中。
from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan_resolve_control_ids(project_uid, control_uid) → _resolve_control_ids(ap_uid, control_uid)
Project 改為 join OscalAssessmentPlan:
ap = session.query(OscalAssessmentPlan).filter(OscalAssessmentPlan.uid == ap_uid).first()
if ap is None:
raise NotFound(GrcErrorCode.GRC_AP_NOT_FOUND)
# 從 AP mapping 取 project_id
mapping = session.query(ProjectAssessmentPlanMapping).filter(
ProjectAssessmentPlanMapping.assessment_plan_id == ap.id
).first()
control_row = session.query(
OscalAssessmentPlanControl.id.label("control_id"),
OscalAssessmentPlanControl.group_id.label("group_id"),
).filter(
OscalAssessmentPlanControl.assessment_plan_id == ap.id,
OscalAssessmentPlanControl.uid == control_uid,
).first()
if control_row is None:
raise NotFound(GrcErrorCode.GRC_CONTROL_NOT_FOUND)
return mapping.project_id, control_row.control_id, control_row.group_idadd_review(project_uid, ...) → add_review(ap_uid, ...)remove_review(project_uid, ...) → remove_review(ap_uid, ...)project_uid 的 review 查詢方法(如 batch fetch)也要更新Dashboard 不接收 ap_uid,但需要處理一個 project 有多個 AP 的情況。
改動策略:在 JOIN 時取每個 project 最新的 AP(created_at 最大的那筆)。
# 建立子查詢:每個 project 最新的 AP mapping
from sqlalchemy import func
latest_ap_sq = (
session.query(
ProjectAssessmentPlanMapping.project_id,
func.max(ProjectAssessmentPlanMapping.id).label("max_mapping_id"),
)
.group_by(ProjectAssessmentPlanMapping.project_id)
.subquery()
)
# 主查詢 JOIN 改為:
.join(latest_ap_sq, latest_ap_sq.c.project_id == Project.id)
.join(
ProjectAssessmentPlanMapping,
ProjectAssessmentPlanMapping.id == latest_ap_sq.c.max_mapping_id,
)確認所有用到 ProjectAssessmentPlanMapping.assessment_plan_id 的 subquery 都正確關聯到最新的那筆 mapping。
Files (Domain):
domain/grc/service/grc_control_group_domain_service.pydomain/grc/service/grc_control_domain_service.pydomain/grc/service/grc_assessment_object_domain_service.pydomain/grc/service/grc_task_setup_domain_service.pydomain/grc/service/grc_review_domain_service.pyFiles (App):
app/grc/service/control_group_service.pyapp/grc/service/control_service.pyapp/grc/service/assessment_object_service.pyapp/grc/service/task_setup_service.pyapp/grc/service/review_service.py所有 domain service 和 app service 的改動都是純粹的參數重新命名(project_uid → ap_uid),因為它們都是 thin delegation。
每個 domain service 的改動模式一致:
# 舊
def get_control_group(self, project_uid, group_uid):
return self._repo.get_control_group_by_uid(project_uid, group_uid)
# 新
def get_control_group(self, ap_uid, group_uid):
return self._repo.get_control_group_by_uid(ap_uid, group_uid)逐一修改:
grc_control_group_domain_service.py:get_control_group()、get_control_groups_and_pager()grc_control_domain_service.py:get_control()、get_controls_and_pager()、get_control_with_ao_detail()grc_assessment_object_domain_service.py:get_assessment_objects_and_pager()grc_task_setup_domain_service.py:get_task_setup_tree()grc_review_domain_service.py:add_review()、remove_review()同樣模式:
# 舊
def list_control_groups(self, project_id, pager, sorts, **filters):
result = self._domain_service.get_control_groups_and_pager(project_id, ...)
# 新
def list_control_groups(self, ap_uid, pager, sorts, **filters):
result = self._domain_service.get_control_groups_and_pager(ap_uid, ...)逐一修改:
control_group_service.py:list_control_groups()、get_control_group()control_service.py:list_controls()、get_control()、get_control_with_ao_detail()assessment_object_service.py:list_assessment_objects()task_setup_service.py:get_task_setup_tree()review_service.py:mark_control()、unmark_control()、mark_ao()、unmark_ao()注意
control_group_service.py: 現有list_control_groups()的第一個參數叫project_id(不是project_uid),但實際傳的是project_uid字串。改為ap_uid時一併修正命名。
Files:
api/grc/routes/control_group_route.pyapi/grc/routes/control_route.pyapi/grc/routes/assessment_object_route.pyapi/grc/routes/job_route.pyapi/grc/routes/task_setup_route.pyapi/grc/routes/review_route.pyapi/grc/__init__.py# ProjectControlGroupDetailResource.get()
# 舊:def get(self, project_uid: str, group_uid: str, ...):
# 新:
def get(self, project_uid: str, ap_uid: str, group_uid: str, ...):
result = control_group_service.get_control_group(ap_uid, group_uid)
# ProjectControlGroupListResource.post()
# 舊:def post(self, project_id: str, ...):
# 新:
def post(self, project_uid: str, ap_uid: str, ...):
result = control_group_service.list_control_groups(ap_uid, pager, sorts, **filters)3 個 Resource class,都加 ap_uid: str 參數:
# ProjectControlListResource.post()
def post(self, project_uid: str, ap_uid: str, group_uid: str, ...):
result = control_service.list_controls(ap_uid, group_uid, pager, sorts, **filters)
# ProjectControlDetailResource.get()
def get(self, project_uid: str, ap_uid: str, group_uid: str, control_uid: str, ...):
result = control_service.get_control(ap_uid, group_uid, control_uid)
# ProjectControlWithAoDetailResource.get()
def get(self, project_uid: str, ap_uid: str, group_uid: str, control_uid: str, ...):
result = control_service.get_control_with_ao_detail(ap_uid, group_uid, control_uid)# ProjectAOListResource.post()
def post(self, project_uid: str, ap_uid: str, group_uid: str, control_uid: str, ...):
result = assessment_object_service.list_assessment_objects(ap_uid, control_uid, pager, sorts, **filters)只改 ProjectJobListResource 和 ProjectJobCreateResource,不改 ProjectJobDetailResource 和 MyJobListResource。
# ProjectJobListResource.post() — 加 ap_uid 但不傳到 service(job repo 用 ao_uid 直查)
def post(self, project_uid: str, ap_uid: str, group_uid: str, control_uid: str, ao_uid: str, ...):
# ap_uid 不傳入 service,只是 URL 結構一致性
result = job_service.list_jobs(ao_uid, pager, sorts, **filters)
# ProjectJobCreateResource.post()
def post(self, project_uid: str, ap_uid: str, ao_uid: str, ...):
result = job_service.create_job(project_uid, ao_uid, ...)Job repo 不需要改查詢邏輯:
list_jobs()和create_job()都是直接用ao_uid查 workflow,不經過ProjectAssessmentPlanMapping。URL 加ap_uid只是為了路徑結構一致。
# ProjectTaskSetupTreeResource.get()
def get(self, project_uid: str, ap_uid: str, ...):
dto = task_setup_service.get_task_setup_tree(ap_uid)4 個 handler method,都加 ap_uid: str:
# ControlReviewResource.post()
def post(self, project_uid: str, ap_uid: str, control_uid: str, ...):
dto = review_service.mark_control(ap_uid, control_uid, user.id)
# ControlReviewResource.delete()
def delete(self, project_uid: str, ap_uid: str, control_uid: str, ...):
dto = review_service.unmark_control(ap_uid, control_uid, user.id)
# AoReviewResource.post()
def post(self, project_uid: str, ap_uid: str, control_uid: str, ao_uid: str, ...):
dto = review_service.mark_ao(ap_uid, control_uid, ao_uid, user.id)
# AoReviewResource.delete()
def delete(self, project_uid: str, ap_uid: str, control_uid: str, ao_uid: str, ...):
dto = review_service.unmark_ao(ap_uid, control_uid, ao_uid, user.id)# 舊 → 新
# Control Group
"/project/<project_uid>/control-group/<group_uid>"
→ "/project/<project_uid>/ap/<ap_uid>/control-group/<group_uid>"
"/project/<project_id>/control-groups/list"
→ "/project/<project_uid>/ap/<ap_uid>/control-groups/list"
# Control
"/project/<project_uid>/control-group/<group_uid>/controls/list"
→ "/project/<project_uid>/ap/<ap_uid>/control-group/<group_uid>/controls/list"
"/project/<project_uid>/control-group/<group_uid>/control/<control_uid>"
→ "/project/<project_uid>/ap/<ap_uid>/control-group/<group_uid>/control/<control_uid>"
"/project/<project_uid>/control-group/<group_uid>/control/<control_uid>/detail"
→ "/project/<project_uid>/ap/<ap_uid>/control-group/<group_uid>/control/<control_uid>/detail"
# Assessment Object
"/project/<project_uid>/control-group/<group_uid>/control/<control_uid>/assessment-objects/list"
→ "/project/<project_uid>/ap/<ap_uid>/control-group/<group_uid>/control/<control_uid>/assessment-objects/list"
# Job
"/project/<project_uid>/control-group/<group_uid>/control/<control_uid>/assessment-object/<ao_uid>/jobs/list"
→ "/project/<project_uid>/ap/<ap_uid>/control-group/<group_uid>/control/<control_uid>/assessment-object/<ao_uid>/jobs/list"
"/project/<project_uid>/assessment-object/<ao_uid>/jobs"
→ "/project/<project_uid>/ap/<ap_uid>/assessment-object/<ao_uid>/jobs"
# Task Setup
"/project/<project_uid>/task-setup/tree"
→ "/project/<project_uid>/ap/<ap_uid>/task-setup/tree"
# Review
"/project/<project_uid>/control/<control_uid>/review"
→ "/project/<project_uid>/ap/<ap_uid>/control/<control_uid>/review"
"/project/<project_uid>/control/<control_uid>/ao/<ao_uid>/review"
→ "/project/<project_uid>/ap/<ap_uid>/control/<control_uid>/ao/<ao_uid>/review"注意:
ProjectControlGroupListResource的 URL 中參數名從project_id改為project_uid(修正命名不一致)。
python main_app.py確認 server 正常啟動,Swagger UI 可見新增的 AP menu endpoint 和更新的 URL。
curl -X GET http://localhost:8000/api/1.0/grc/project/<existing_project_uid>/assessment-plans/menu \
-H "Authorization: Bearer <token>"預期回傳該專案的 AP 列表(現有專案應有 1 筆)。
curl -X POST http://localhost:8000/api/1.0/grc/project/<project_uid>/ap/<ap_uid>/control-groups/list \
-H "Authorization: Bearer <token>" \
-H "Content-Type: application/json" \
-d '{"pager": {"page": 1, "page_size": 10}}'預期回傳與舊 URL 相同的資料。
curl -X GET http://localhost:8000/api/1.0/grc/project/<project_uid>/ap/<ap_uid>/task-setup/tree \
-H "Authorization: Bearer <token>"curl -X GET http://localhost:8000/api/1.0/grc/dashboard/summary \
-H "Authorization: Bearer <token>"確認回傳正常,不受多 AP 影響。