Stage 1:多 AP 基礎建設 Implementation Plan

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


§1

File Structure

新增檔案

檔案 職責
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_uidap_uid
domain/grc/service/grc_control_domain_service.py 參數 project_uidap_uid
domain/grc/service/grc_assessment_object_domain_service.py 參數 project_uidap_uid
domain/grc/service/grc_task_setup_domain_service.py 參數 project_uidap_uid
domain/grc/service/grc_review_domain_service.py 參數 project_uidap_uid
domain/grc/service/grc_project_domain_service.py 新增 get_assessment_plans_menu()
app/grc/service/control_group_service.py 參數 project_uid/project_idap_uid
app/grc/service/control_service.py 參數 project_uidap_uid
app/grc/service/assessment_object_service.py 參數 project_uidap_uid
app/grc/service/task_setup_service.py 參數 project_uidap_uid
app/grc/service/review_service.py 參數 project_uidap_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

§2

Task 1: Foundation — Error Code + New Files

Files:

  • Modify: common/code/grc_error_code.py
  • Create: app/grc/dto/assessment_plan_dto.py
  • Create: api/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)

§3

Task 2: Repo 層 — 控制項結構查詢改用 ap_uid

Files:

  • Modify: infra/grc/repository/grc_control_group_repo_impl.py
  • Modify: infra/grc/repository/grc_control_repo_impl.py
  • Modify: infra/grc/repository/grc_assessment_object_repo_impl.py
  • Modify: infra/grc/repository/grc_task_setup_repo_impl.py

所有 4 個 repo 的改動模式一致:把 project_uid → ProjectAssessmentPlanMapping.first() 替換為 ap_uid → OscalAssessmentPlan

共通改動模式

每個 repo 需要:

  1. 新增 import:from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan
  2. 找到所有透過 ProjectAssessmentPlanMapping + Project.uid == project_uid 解析 ap_id 的查詢區塊
  3. 替換為直接用 ap_uidOscalAssessmentPlan

舊模式:

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 反查。

  1. 新增 import:from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan
  2. get_control_group_by_uid(project_uid, group_uid) → 改簽名為 get_control_group_by_uid(ap_uid, group_uid)
    • 替換 ap_row 查詢區塊(約 line 58-66)
    • 如果 method 內有用 project_id(participant 查詢),改從 ProjectAssessmentPlanMapping 反查:
      mapping = session.query(ProjectAssessmentPlanMapping).filter(
          ProjectAssessmentPlanMapping.assessment_plan_id == ap_id
      ).first()
      project_id = mapping.project_id if mapping else None
  3. list_control_groups(project_uid, pager, sorts, <strong>filters) → 改簽名為 list_control_groups(ap_uid, pager, sorts, </strong>filters)
    • 替換 ap_row 查詢區塊(約 line 246-254)
    • 同上處理 project_id
  1. 新增 import:from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan
  2. get_control_by_uid(project_uid, group_uid, control_uid)get_control_by_uid(ap_uid, group_uid, control_uid)
    • 替換 ap_row 查詢區塊(約 line 68-76)
  3. list_controls(project_uid, group_uid, pager, sorts, **filters)list_controls(ap_uid, group_uid, pager, sorts, **filters)
    • 替換 ap_row 查詢區塊
  4. get_control_with_ao_detail(project_uid, group_uid, control_uid)get_control_with_ao_detail(ap_uid, group_uid, control_uid)
    • 替換 ap_row 查詢區塊
    • 此方法內有查 SSP(透過 AP 的 ssp_id),確認 ap 物件的 ssp_id 屬性可用
  1. 新增 import:from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan
  2. list_assessment_objects(project_uid, control_uid, pager, sorts, **filters)list_assessment_objects(ap_uid, control_uid, pager, sorts, **filters)
    • 替換 ap_row 查詢區塊(約 line 52-61)
    • 此 repo 只需要 ap_id,不需要 project_id
  1. 新增 import:from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan
  2. get_task_setup_tree(project_uid)get_task_setup_tree(ap_uid)
    • 替換 ap_row 查詢區塊(約 line 48-57)
    • 此方法需要 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_name

§4

Task 3: Repo 層 — Review + Dashboard

Files:

  • Modify: infra/grc/repository/grc_review_repo_impl.py
  • Modify: infra/grc/repository/grc_dashboard_repo_impl.py

Review repo 的 AP 解析在 helper method _resolve_control_ids() 中。

  1. 新增 import:from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan
  2. _resolve_control_ids(project_uid, control_uid)_resolve_control_ids(ap_uid, control_uid)
    • 替換查詢(約 line 31-50),從 join 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_id
  3. add_review(project_uid, ...)add_review(ap_uid, ...)
  4. remove_review(project_uid, ...)remove_review(ap_uid, ...)
  5. 其他用到 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。


§5

Task 4: Domain Service + App Service 層

Files (Domain):

  • Modify: domain/grc/service/grc_control_group_domain_service.py
  • Modify: domain/grc/service/grc_control_domain_service.py
  • Modify: domain/grc/service/grc_assessment_object_domain_service.py
  • Modify: domain/grc/service/grc_task_setup_domain_service.py
  • Modify: domain/grc/service/grc_review_domain_service.py

Files (App):

  • Modify: app/grc/service/control_group_service.py
  • Modify: app/grc/service/control_service.py
  • Modify: app/grc/service/assessment_object_service.py
  • Modify: app/grc/service/task_setup_service.py
  • Modify: app/grc/service/review_service.py

所有 domain service 和 app service 的改動都是純粹的參數重新命名(project_uidap_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.pyget_control_group()get_control_groups_and_pager()
  • grc_control_domain_service.pyget_control()get_controls_and_pager()get_control_with_ao_detail()
  • grc_assessment_object_domain_service.pyget_assessment_objects_and_pager()
  • grc_task_setup_domain_service.pyget_task_setup_tree()
  • grc_review_domain_service.pyadd_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.pylist_control_groups()get_control_group()
  • control_service.pylist_controls()get_control()get_control_with_ao_detail()
  • assessment_object_service.pylist_assessment_objects()
  • task_setup_service.pyget_task_setup_tree()
  • review_service.pymark_control()unmark_control()mark_ao()unmark_ao()

注意 control_group_service.py 現有 list_control_groups() 的第一個參數叫 project_id(不是 project_uid),但實際傳的是 project_uid 字串。改為 ap_uid 時一併修正命名。


§6

Task 5: Route 層 + Blueprint

Files:

  • Modify: api/grc/routes/control_group_route.py
  • Modify: api/grc/routes/control_route.py
  • Modify: api/grc/routes/assessment_object_route.py
  • Modify: api/grc/routes/job_route.py
  • Modify: api/grc/routes/task_setup_route.py
  • Modify: api/grc/routes/review_route.py
  • Modify: api/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)

只改 ProjectJobListResourceProjectJobCreateResource,不改 ProjectJobDetailResourceMyJobListResource

# 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(修正命名不一致)。


§7

Task 6: 新增 AP Menu API

Files:

  • Create: api/grc/routes/assessment_plan_route.py
  • Modify: infra/grc/repository/grc_project_repo_impl.py
  • Modify: domain/grc/service/grc_project_domain_service.py
  • Modify: app/grc/service/project_service.py
  • Modify: api/grc/__init__.py
def get_assessment_plans_menu(self, project_uid: str) -> list:
    from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan

    session = get_session()
    rows = (
        session.query(OscalAssessmentPlan)
        .join(
            ProjectAssessmentPlanMapping,
            ProjectAssessmentPlanMapping.assessment_plan_id == OscalAssessmentPlan.id,
        )
        .join(Project, Project.id == ProjectAssessmentPlanMapping.project_id)
        .filter(Project.uid == project_uid)
        .order_by(OscalAssessmentPlan.created_at.asc())
        .all()
    )
    return rows
def get_assessment_plans_menu(self, project_uid: str) -> list:
    return self._repo.get_assessment_plans_menu(project_uid)
from app.grc.dto.assessment_plan_dto import GrcAssessmentPlanMenuDto

@transaction
def get_assessment_plans_menu(self, project_uid: str) -> list[GrcAssessmentPlanMenuDto]:
    rows = self._domain_service.get_assessment_plans_menu(project_uid)
    return [
        GrcAssessmentPlanMenuDto.from_row(row, round_number=idx + 1)
        for idx, row in enumerate(rows)
    ]
import logging

from dependency_injector.wiring import inject, Provide
from flask_apispec import MethodResource, doc, marshal_with
from flask_jwt_extended import jwt_required

from api.grc.serializers.assessment_plan import AssessmentPlanMenuResponseSchema
from app.grc.service.project_service import ProjectService
from common.enum.schema_code import AUTH_PARAMS
from common.util.response_util import return_response
from di_containers.containers import Containers

logger = logging.getLogger(__name__)


class AssessmentPlanMenuResource(MethodResource):
    """GRC 專案稽核輪次(AP)選單"""

    @doc(
        description="取得指定專案的 Assessment Plan 列表(選擇稽核輪次)",
        tags=["GRC Projects"],
        params=AUTH_PARAMS,
    )
    @marshal_with(AssessmentPlanMenuResponseSchema(many=True), apply=False)
    @jwt_required()
    @inject
    def get(
        self,
        project_uid: str,
        project_service: ProjectService = Provide[Containers.grc_container.project_service],
    ):
        result = project_service.get_assessment_plans_menu(project_uid)
        return return_response(True, AssessmentPlanMenuResponseSchema(many=True).dump(result))

在 imports 加:

from api.grc.routes.assessment_plan_route import AssessmentPlanMenuResource

api.add_resource 區塊加:

api.add_resource(AssessmentPlanMenuResource, "/project/<project_uid>/assessment-plans/menu")

§8

Task 7: 驗證

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 影響。