init commit

This commit is contained in:
2026-03-18 06:40:53 -05:00
commit dd227be9b0
53 changed files with 18575 additions and 0 deletions

0
backend/app/__init__.py Normal file
View File

49
backend/app/auth.py Normal file
View File

@@ -0,0 +1,49 @@
import os
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from pydantic import BaseModel
PASSWORD = os.environ.get("REQ_PLANNER_PASSWORD", "what_is_this_2")
SECRET_KEY = os.environ.get("REQ_PLANNER_SECRET", os.urandom(32).hex())
ALGORITHM = "HS256"
TOKEN_EXPIRE_HOURS = 24
security = HTTPBearer()
router = APIRouter(prefix="/api/auth", tags=["auth"])
class LoginRequest(BaseModel):
password: str
class TokenResponse(BaseModel):
token: str
def create_token() -> str:
expire = datetime.now(timezone.utc) + timedelta(hours=TOKEN_EXPIRE_HOURS)
return jwt.encode({"exp": expire}, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> bool:
try:
jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
return True
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
)
@router.post("/login", response_model=TokenResponse)
def login(data: LoginRequest):
if data.password != PASSWORD:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid password",
)
return TokenResponse(token=create_token())

22
backend/app/database.py Normal file
View File

@@ -0,0 +1,22 @@
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
DATA_DIR = Path(__file__).resolve().parent.parent.parent / "data"
DATA_DIR.mkdir(exist_ok=True)
DB_PATH = DATA_DIR / "planner.db"
engine = create_engine(f"sqlite:///{DB_PATH}", echo=False)
SessionLocal = sessionmaker(bind=engine, class_=Session, expire_on_commit=False)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

34
backend/app/main.py Normal file
View File

@@ -0,0 +1,34 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .auth import router as auth_router
from .database import Base, engine
from .routers import layouts, links, projects, requirements, stats
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Requirements Planner",
description="General-purpose requirements management with quality gates",
version="0.1.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth_router)
app.include_router(requirements.router)
app.include_router(links.router)
app.include_router(projects.router)
app.include_router(stats.router)
app.include_router(layouts.router)
@app.get("/api/health")
def health():
return {"status": "ok"}

221
backend/app/models.py Normal file
View File

@@ -0,0 +1,221 @@
import enum
from datetime import datetime, timezone
from sqlalchemy import Enum, ForeignKey, Index, Integer, String, Text, Boolean, DateTime, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .database import Base
class Priority(str, enum.Enum):
critical = "critical"
high = "high"
medium = "medium"
low = "low"
class Status(str, enum.Enum):
idea = "idea"
draft = "draft"
todo = "todo"
in_progress = "in_progress"
done = "done"
class NodeType(str, enum.Enum):
pillar = "pillar"
requirement = "requirement"
sub_requirement = "sub_requirement"
leaf = "leaf"
planning = "planning"
class CodeLanguage(str, enum.Enum):
python = "python"
cpp = "cpp"
typescript = "typescript"
class TestStatus(str, enum.Enum):
not_written = "not_written"
written = "written"
passing = "passing"
failing = "failing"
def _utcnow():
return datetime.now(timezone.utc)
class RequirementNode(Base):
__tablename__ = "requirement_nodes"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
title: Mapped[str] = mapped_column(String(256), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
priority: Mapped[Priority] = mapped_column(Enum(Priority), default=Priority.medium)
phase: Mapped[int | None] = mapped_column(Integer, nullable=True)
status: Mapped[Status] = mapped_column(Enum(Status), default=Status.draft)
node_type: Mapped[NodeType] = mapped_column(Enum(NodeType), default=NodeType.requirement)
parent_id: Mapped[str | None] = mapped_column(
String(64), ForeignKey("requirement_nodes.id", ondelete="CASCADE"), nullable=True
)
sort_order: Mapped[int] = mapped_column(Integer, default=0)
project_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("projects.id", ondelete="CASCADE"), nullable=True
)
# Quality gate flags
traceable: Mapped[bool] = mapped_column(Boolean, default=False)
measurable: Mapped[bool] = mapped_column(Boolean, default=False)
consistent: Mapped[bool] = mapped_column(Boolean, default=False)
single_purpose: Mapped[bool] = mapped_column(Boolean, default=False)
# Issue tracker integration
gitea_issue_number: Mapped[int | None] = mapped_column(Integer, nullable=True)
gitea_repo: Mapped[str | None] = mapped_column(String(256), nullable=True)
# Acceptance criteria and code (all node types)
acceptance_criteria: Mapped[str | None] = mapped_column(Text, nullable=True)
test_spec: Mapped[str | None] = mapped_column(Text, nullable=True)
test_code: Mapped[str | None] = mapped_column(Text, nullable=True)
impl_code: Mapped[str | None] = mapped_column(Text, nullable=True)
code_language: Mapped[CodeLanguage | None] = mapped_column(
Enum(CodeLanguage), nullable=True
)
target_test_file: Mapped[str | None] = mapped_column(String(512), nullable=True)
target_impl_file: Mapped[str | None] = mapped_column(String(512), nullable=True)
test_status: Mapped[TestStatus] = mapped_column(
Enum(TestStatus), default=TestStatus.not_written
)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=_utcnow, onupdate=_utcnow
)
# Relationships
parent: Mapped["RequirementNode | None"] = relationship(
"RequirementNode", remote_side="RequirementNode.id", back_populates="children"
)
children: Mapped[list["RequirementNode"]] = relationship(
"RequirementNode", back_populates="parent", cascade="all, delete-orphan",
order_by="RequirementNode.sort_order"
)
project: Mapped["Project | None"] = relationship("Project", back_populates="requirements")
# Cross-pillar links (outgoing)
outgoing_links: Mapped[list["CrossPillarLink"]] = relationship(
"CrossPillarLink", foreign_keys="CrossPillarLink.source_id",
back_populates="source", cascade="all, delete-orphan"
)
incoming_links: Mapped[list["CrossPillarLink"]] = relationship(
"CrossPillarLink", foreign_keys="CrossPillarLink.target_id",
back_populates="target", cascade="all, delete-orphan"
)
history: Mapped[list["RequirementHistory"]] = relationship(
"RequirementHistory", back_populates="requirement",
cascade="all, delete-orphan", order_by="RequirementHistory.version"
)
class CrossPillarLink(Base):
__tablename__ = "cross_pillar_links"
source_id: Mapped[str] = mapped_column(
String(64), ForeignKey("requirement_nodes.id", ondelete="CASCADE"), primary_key=True
)
target_id: Mapped[str] = mapped_column(
String(64), ForeignKey("requirement_nodes.id", ondelete="CASCADE"), primary_key=True
)
relationship_type: Mapped[str] = mapped_column(
String(64), default="shared_parent"
)
source: Mapped[RequirementNode] = relationship(
"RequirementNode", foreign_keys=[source_id], back_populates="outgoing_links"
)
target: Mapped[RequirementNode] = relationship(
"RequirementNode", foreign_keys=[target_id], back_populates="incoming_links"
)
class Project(Base):
__tablename__ = "projects"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(256), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
repo_path: Mapped[str | None] = mapped_column(String(512), nullable=True)
repo_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
requirements_yaml_path: Mapped[str | None] = mapped_column(String(512), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
requirements: Mapped[list[RequirementNode]] = relationship(
"RequirementNode", back_populates="project", cascade="all, delete-orphan"
)
layouts: Mapped[list["GraphLayout"]] = relationship(
"GraphLayout", back_populates="project", cascade="all, delete-orphan"
)
class RequirementHistory(Base):
__tablename__ = "requirement_history"
__table_args__ = (
UniqueConstraint("requirement_id", "version", name="uq_requirement_version"),
Index("idx_req_history_req_version", "requirement_id", "version"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
requirement_id: Mapped[str] = mapped_column(
String(64), ForeignKey("requirement_nodes.id", ondelete="CASCADE"), nullable=False
)
version: Mapped[int] = mapped_column(Integer, nullable=False)
# Snapshot of editable fields
title: Mapped[str] = mapped_column(String(256), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
priority: Mapped[str] = mapped_column(String(16), nullable=False)
phase: Mapped[int | None] = mapped_column(Integer, nullable=True)
status: Mapped[str] = mapped_column(String(16), nullable=False)
node_type: Mapped[str] = mapped_column(String(32), nullable=False)
acceptance_criteria: Mapped[str | None] = mapped_column(Text, nullable=True)
test_spec: Mapped[str | None] = mapped_column(Text, nullable=True)
test_code: Mapped[str | None] = mapped_column(Text, nullable=True)
impl_code: Mapped[str | None] = mapped_column(Text, nullable=True)
code_language: Mapped[str | None] = mapped_column(String(16), nullable=True)
target_test_file: Mapped[str | None] = mapped_column(String(512), nullable=True)
target_impl_file: Mapped[str | None] = mapped_column(String(512), nullable=True)
measurable: Mapped[bool] = mapped_column(Boolean, default=False)
consistent: Mapped[bool] = mapped_column(Boolean, default=False)
single_purpose: Mapped[bool] = mapped_column(Boolean, default=False)
gitea_issue_number: Mapped[int | None] = mapped_column(Integer, nullable=True)
gitea_repo: Mapped[str | None] = mapped_column(String(256), nullable=True)
# Metadata
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
change_summary: Mapped[str | None] = mapped_column(String(512), nullable=True)
requirement: Mapped[RequirementNode] = relationship(
"RequirementNode", back_populates="history"
)
class GraphLayout(Base):
__tablename__ = "graph_layouts"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(128), nullable=False)
description: Mapped[str | None] = mapped_column(String(512), nullable=True)
project_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("projects.id", ondelete="CASCADE"), nullable=True
)
node_positions: Mapped[str] = mapped_column(Text, nullable=False)
viewport: Mapped[str] = mapped_column(Text, nullable=False)
is_default: Mapped[bool] = mapped_column(Boolean, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=_utcnow, onupdate=_utcnow
)
project: Mapped["Project | None"] = relationship("Project", back_populates="layouts")

54
backend/app/quality.py Normal file
View File

@@ -0,0 +1,54 @@
from .models import RequirementNode, NodeType
from .schemas import QualityReport
def validate_requirement(node: RequirementNode, children: list[RequirementNode]) -> QualityReport:
"""Run quality gates on a requirement node.
Quality gates:
1. Traceable — auto-calculated: has parent (non-pillar) + description
2. Measurable — user toggle (manually set by the user)
3. Consistent — user toggle (manually set by the user)
4. Single-purpose — user toggle (manually set by the user)
"""
errors: list[str] = []
warnings: list[str] = []
# 1. Traceable — auto-calculated
traceable = True
if node.node_type != NodeType.pillar:
if not node.parent_id:
errors.append("Non-pillar node must have a parent (traceability)")
traceable = False
if not node.description:
warnings.append("Missing description — hard to trace how this serves parent")
else:
if not node.description:
warnings.append("Pillar should have a description explaining its purpose")
# 2. Measurable — user toggle, but forced false without test_spec
measurable = bool(node.measurable)
if not node.acceptance_criteria:
warnings.append("Missing acceptance criteria")
if not node.test_spec:
errors.append("Missing test specification (GIVEN/WHEN/THEN) — cannot be measurable")
measurable = False
# 3. Consistent — user toggle
consistent = bool(node.consistent)
# 4. Single-purpose — user toggle, but warn on heuristic
single_purpose = bool(node.single_purpose)
title_lower = node.title.lower()
if " and " in title_lower and " and " not in title_lower.split("(")[0]:
warnings.append("Title contains 'and' — may serve multiple purposes")
return QualityReport(
traceable=traceable,
measurable=measurable,
consistent=consistent,
single_purpose=single_purpose,
errors=errors,
warnings=warnings,
valid=len(errors) == 0,
)

View File

View File

@@ -0,0 +1,102 @@
import json
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..auth import verify_token
from ..database import get_db
from ..models import GraphLayout
from ..schemas import (
GraphLayoutCreate,
GraphLayoutDetailResponse,
GraphLayoutResponse,
GraphLayoutUpdate,
)
router = APIRouter(prefix="/api/layouts", tags=["layouts"], dependencies=[Depends(verify_token)])
def _to_response(layout: GraphLayout) -> GraphLayoutResponse:
return GraphLayoutResponse.model_validate(layout)
def _to_detail_response(layout: GraphLayout) -> GraphLayoutDetailResponse:
base = GraphLayoutResponse.model_validate(layout)
return GraphLayoutDetailResponse(
**base.model_dump(),
node_positions=json.loads(layout.node_positions),
viewport=json.loads(layout.viewport),
)
@router.get("", response_model=list[GraphLayoutResponse])
def list_layouts(project_id: int | None = None, db: Session = Depends(get_db)):
query = db.query(GraphLayout)
if project_id is not None:
query = query.filter(GraphLayout.project_id == project_id)
return [_to_response(l) for l in query.order_by(GraphLayout.updated_at.desc()).all()]
@router.get("/{layout_id}", response_model=GraphLayoutDetailResponse)
def get_layout(layout_id: int, db: Session = Depends(get_db)):
layout = db.get(GraphLayout, layout_id)
if not layout:
raise HTTPException(404, "Layout not found")
return _to_detail_response(layout)
@router.post("", response_model=GraphLayoutDetailResponse, status_code=201)
def create_layout(data: GraphLayoutCreate, db: Session = Depends(get_db)):
if data.is_default:
db.query(GraphLayout).filter(
GraphLayout.project_id == data.project_id
).update({"is_default": False})
layout = GraphLayout(
name=data.name,
description=data.description,
project_id=data.project_id,
node_positions=json.dumps(data.node_positions),
viewport=json.dumps(data.viewport),
is_default=data.is_default,
)
db.add(layout)
db.commit()
db.refresh(layout)
return _to_detail_response(layout)
@router.put("/{layout_id}", response_model=GraphLayoutDetailResponse)
def update_layout(layout_id: int, data: GraphLayoutUpdate, db: Session = Depends(get_db)):
layout = db.get(GraphLayout, layout_id)
if not layout:
raise HTTPException(404, "Layout not found")
update_data = data.model_dump(exclude_unset=True)
if update_data.get("is_default"):
db.query(GraphLayout).filter(
GraphLayout.project_id == layout.project_id,
GraphLayout.id != layout_id,
).update({"is_default": False})
if "node_positions" in update_data:
update_data["node_positions"] = json.dumps(update_data["node_positions"])
if "viewport" in update_data:
update_data["viewport"] = json.dumps(update_data["viewport"])
for key, value in update_data.items():
setattr(layout, key, value)
db.commit()
db.refresh(layout)
return _to_detail_response(layout)
@router.delete("/{layout_id}", status_code=204)
def delete_layout(layout_id: int, db: Session = Depends(get_db)):
layout = db.get(GraphLayout, layout_id)
if not layout:
raise HTTPException(404, "Layout not found")
db.delete(layout)
db.commit()

View File

@@ -0,0 +1,55 @@
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import or_
from sqlalchemy.orm import Session
from ..auth import verify_token
from ..database import get_db
from ..models import CrossPillarLink, RequirementNode
from ..schemas import LinkCreate, LinkResponse
router = APIRouter(prefix="/api/links", tags=["links"], dependencies=[Depends(verify_token)])
@router.get("", response_model=list[LinkResponse])
def list_links(
node_id: Optional[str] = Query(None, description="Filter links involving this node"),
db: Session = Depends(get_db),
):
q = db.query(CrossPillarLink)
if node_id:
q = q.filter(
or_(
CrossPillarLink.source_id == node_id,
CrossPillarLink.target_id == node_id,
)
)
return q.all()
@router.post("", response_model=LinkResponse, status_code=201)
def create_link(data: LinkCreate, db: Session = Depends(get_db)):
if not db.get(RequirementNode, data.source_id):
raise HTTPException(404, f"Source {data.source_id} not found")
if not db.get(RequirementNode, data.target_id):
raise HTTPException(404, f"Target {data.target_id} not found")
existing = db.get(CrossPillarLink, (data.source_id, data.target_id))
if existing:
raise HTTPException(409, "Link already exists")
link = CrossPillarLink(**data.model_dump())
db.add(link)
db.commit()
db.refresh(link)
return link
@router.delete("/{source_id}/{target_id}", status_code=204)
def delete_link(source_id: str, target_id: str, db: Session = Depends(get_db)):
link = db.get(CrossPillarLink, (source_id, target_id))
if not link:
raise HTTPException(404, "Link not found")
db.delete(link)
db.commit()

View File

@@ -0,0 +1,40 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..auth import verify_token
from ..database import get_db
from ..models import Project
from ..schemas import ProjectCreate, ProjectResponse
router = APIRouter(prefix="/api/projects", tags=["projects"], dependencies=[Depends(verify_token)])
@router.get("", response_model=list[ProjectResponse])
def list_projects(db: Session = Depends(get_db)):
return db.query(Project).all()
@router.post("", response_model=ProjectResponse, status_code=201)
def create_project(data: ProjectCreate, db: Session = Depends(get_db)):
project = Project(**data.model_dump())
db.add(project)
db.commit()
db.refresh(project)
return project
@router.get("/{project_id}", response_model=ProjectResponse)
def get_project(project_id: int, db: Session = Depends(get_db)):
project = db.get(Project, project_id)
if not project:
raise HTTPException(404, f"Project {project_id} not found")
return project
@router.delete("/{project_id}", status_code=204)
def delete_project(project_id: int, db: Session = Depends(get_db)):
project = db.get(Project, project_id)
if not project:
raise HTTPException(404, f"Project {project_id} not found")
db.delete(project)
db.commit()

View File

@@ -0,0 +1,252 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import func
from sqlalchemy.orm import Session
from ..auth import verify_token
from ..database import get_db
from ..models import RequirementNode, RequirementHistory
from ..quality import validate_requirement
from ..schemas import (
RequirementCreate,
RequirementResponse,
RequirementTreeNode,
RequirementUpdate,
QualityReport,
RequirementHistorySummary,
RequirementHistoryDetail,
)
router = APIRouter(prefix="/api/requirements", tags=["requirements"], dependencies=[Depends(verify_token)])
# ─── Snapshot fields to copy between RequirementNode and RequirementHistory ───
_SNAPSHOT_FIELDS = [
"title", "description", "priority", "phase", "status", "node_type",
"acceptance_criteria", "test_spec", "test_code", "impl_code",
"code_language", "target_test_file", "target_impl_file",
"measurable", "consistent", "single_purpose",
"gitea_issue_number", "gitea_repo",
]
def _create_history_entry(
db: Session, node: RequirementNode, version: int, summary: str | None = None
) -> RequirementHistory:
data = {field: getattr(node, field) for field in _SNAPSHOT_FIELDS}
# Convert enums to their string values for plain String columns
for key in ("priority", "status", "node_type", "code_language"):
val = data.get(key)
if val is not None and hasattr(val, "value"):
data[key] = val.value
entry = RequirementHistory(
requirement_id=node.id,
version=version,
change_summary=summary,
**data,
)
db.add(entry)
return entry
def _next_version(db: Session, req_id: str) -> int:
max_ver = db.query(func.max(RequirementHistory.version)).filter(
RequirementHistory.requirement_id == req_id
).scalar()
return (max_ver or 0) + 1
def _auto_summary(update_data: dict) -> str:
fields = [k for k in update_data if k in _SNAPSHOT_FIELDS]
if not fields:
return "Update"
return "Updated " + ", ".join(fields)
def _to_response(node: RequirementNode, include_children: bool = False) -> RequirementResponse:
quality = validate_requirement(node, node.children)
resp = RequirementResponse.model_validate(node)
resp.quality = quality
if include_children:
resp.children = [_to_response(c) for c in node.children]
return resp
def _to_tree_node(node: RequirementNode) -> RequirementTreeNode:
quality = validate_requirement(node, node.children)
return RequirementTreeNode(
id=node.id,
title=node.title,
priority=node.priority,
phase=node.phase,
status=node.status,
node_type=node.node_type,
parent_id=node.parent_id,
child_count=len(node.children),
traceable=quality.traceable,
measurable=quality.measurable,
consistent=node.consistent,
single_purpose=node.single_purpose,
gitea_issue_number=node.gitea_issue_number,
children=[_to_tree_node(c) for c in node.children],
)
@router.get("", response_model=list[RequirementResponse])
def list_requirements(
project_id: int | None = None,
parent_id: str | None = None,
db: Session = Depends(get_db),
):
query = db.query(RequirementNode)
if project_id is not None:
query = query.filter(RequirementNode.project_id == project_id)
if parent_id is not None:
query = query.filter(RequirementNode.parent_id == parent_id)
nodes = query.order_by(RequirementNode.sort_order).all()
return [_to_response(n) for n in nodes]
@router.get("/tree", response_model=list[RequirementTreeNode])
def get_tree(project_id: int | None = None, db: Session = Depends(get_db)):
query = db.query(RequirementNode).filter(RequirementNode.parent_id.is_(None))
if project_id is not None:
query = query.filter(RequirementNode.project_id == project_id)
roots = query.order_by(RequirementNode.sort_order).all()
return [_to_tree_node(r) for r in roots]
@router.get("/{req_id}", response_model=RequirementResponse)
def get_requirement(req_id: str, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
return _to_response(node, include_children=True)
@router.post("", response_model=RequirementResponse, status_code=201)
def create_requirement(data: RequirementCreate, db: Session = Depends(get_db)):
if db.get(RequirementNode, data.id):
raise HTTPException(409, f"Requirement {data.id} already exists")
if data.parent_id and not db.get(RequirementNode, data.parent_id):
raise HTTPException(404, f"Parent {data.parent_id} not found")
node = RequirementNode(**data.model_dump())
# Auto-compute traceable flag only
quality = validate_requirement(node, [])
node.traceable = quality.traceable
db.add(node)
db.flush()
_create_history_entry(db, node, version=1, summary="Initial version")
db.commit()
db.refresh(node)
return _to_response(node)
@router.put("/{req_id}", response_model=RequirementResponse)
def update_requirement(req_id: str, data: RequirementUpdate, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(node, key, value)
# Auto-compute traceable flag only (M, C, S are user-set toggles)
quality = validate_requirement(node, node.children)
node.traceable = quality.traceable
version = _next_version(db, req_id)
_create_history_entry(db, node, version, _auto_summary(update_data))
db.commit()
db.refresh(node)
return _to_response(node, include_children=True)
@router.delete("/{req_id}", status_code=204)
def delete_requirement(req_id: str, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
db.delete(node)
db.commit()
@router.patch("/{req_id}/status")
def update_status(req_id: str, status: str, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
node.status = status
db.commit()
return {"id": req_id, "status": status}
@router.post("/{req_id}/validate", response_model=QualityReport)
def validate_node(req_id: str, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
return validate_requirement(node, node.children)
@router.get("/{req_id}/children", response_model=list[RequirementResponse])
def get_children(req_id: str, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
return [_to_response(c) for c in node.children]
# ─── History Endpoints ────────────────────────────────────────────────────
@router.get("/{req_id}/history", response_model=list[RequirementHistorySummary])
def get_history(req_id: str, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
entries = db.query(RequirementHistory).filter(
RequirementHistory.requirement_id == req_id
).order_by(RequirementHistory.version.desc()).all()
return entries
@router.get("/{req_id}/history/{version}", response_model=RequirementHistoryDetail)
def get_history_version(req_id: str, version: int, db: Session = Depends(get_db)):
entry = db.query(RequirementHistory).filter(
RequirementHistory.requirement_id == req_id,
RequirementHistory.version == version,
).first()
if not entry:
raise HTTPException(404, f"Version {version} not found for {req_id}")
return entry
@router.post("/{req_id}/history/{version}/restore", response_model=RequirementResponse)
def restore_version(req_id: str, version: int, db: Session = Depends(get_db)):
node = db.get(RequirementNode, req_id)
if not node:
raise HTTPException(404, f"Requirement {req_id} not found")
entry = db.query(RequirementHistory).filter(
RequirementHistory.requirement_id == req_id,
RequirementHistory.version == version,
).first()
if not entry:
raise HTTPException(404, f"Version {version} not found for {req_id}")
# Restore fields from history snapshot
for field in _SNAPSHOT_FIELDS:
setattr(node, field, getattr(entry, field))
quality = validate_requirement(node, node.children)
node.traceable = quality.traceable
new_version = _next_version(db, req_id)
_create_history_entry(db, node, new_version, f"Restored from version {version}")
db.commit()
db.refresh(node)
return _to_response(node, include_children=True)

View File

@@ -0,0 +1,55 @@
from fastapi import APIRouter, Depends
from sqlalchemy import func
from sqlalchemy.orm import Session
from ..auth import verify_token
from ..database import get_db
from ..models import NodeType, RequirementNode
from ..schemas import StatsResponse
router = APIRouter(prefix="/api/stats", tags=["stats"], dependencies=[Depends(verify_token)])
@router.get("", response_model=StatsResponse)
def get_stats(project_id: int | None = None, db: Session = Depends(get_db)):
query = db.query(RequirementNode)
if project_id is not None:
query = query.filter(RequirementNode.project_id == project_id)
nodes = query.all()
total = len(nodes)
by_type = {}
by_status: dict[str, int] = {}
by_priority: dict[str, int] = {}
quality_passing = 0
leaves_with_tests = 0
total_leaves = 0
for n in nodes:
by_type[n.node_type.value] = by_type.get(n.node_type.value, 0) + 1
by_status[n.status.value] = by_status.get(n.status.value, 0) + 1
by_priority[n.priority.value] = by_priority.get(n.priority.value, 0) + 1
if n.traceable and n.measurable and n.consistent and n.single_purpose:
quality_passing += 1
if n.node_type == NodeType.leaf:
total_leaves += 1
if n.test_spec:
leaves_with_tests += 1
coverage = (leaves_with_tests / total_leaves * 100) if total_leaves > 0 else 0.0
return StatsResponse(
total_nodes=total,
pillars=by_type.get("pillar", 0),
requirements=by_type.get("requirement", 0),
sub_requirements=by_type.get("sub_requirement", 0),
leaves=by_type.get("leaf", 0),
by_status=by_status,
by_priority=by_priority,
quality_passing=quality_passing,
quality_failing=total - quality_passing,
coverage_percent=round(coverage, 1),
)

229
backend/app/schemas.py Normal file
View File

@@ -0,0 +1,229 @@
from datetime import datetime
from pydantic import BaseModel, Field
from .models import CodeLanguage, NodeType, Priority, Status, TestStatus
# ─── Requirement ─────────────────────────────────────────────────────────
class RequirementCreate(BaseModel):
id: str = Field(..., max_length=64)
title: str = Field(..., max_length=256)
description: str | None = None
priority: Priority = Priority.medium
phase: int | None = None
status: Status = Status.draft
node_type: NodeType = NodeType.requirement
parent_id: str | None = None
sort_order: int = 0
project_id: int | None = None
gitea_issue_number: int | None = None
gitea_repo: str | None = None
acceptance_criteria: str | None = None
test_spec: str | None = None
test_code: str | None = None
impl_code: str | None = None
code_language: CodeLanguage | None = None
target_test_file: str | None = None
target_impl_file: str | None = None
class RequirementUpdate(BaseModel):
title: str | None = None
description: str | None = None
priority: Priority | None = None
phase: int | None = None
status: Status | None = None
node_type: NodeType | None = None
parent_id: str | None = None
sort_order: int | None = None
gitea_issue_number: int | None = None
gitea_repo: str | None = None
acceptance_criteria: str | None = None
test_spec: str | None = None
test_code: str | None = None
impl_code: str | None = None
code_language: CodeLanguage | None = None
target_test_file: str | None = None
target_impl_file: str | None = None
measurable: bool | None = None
consistent: bool | None = None
single_purpose: bool | None = None
class QualityReport(BaseModel):
traceable: bool
measurable: bool
consistent: bool
single_purpose: bool
errors: list[str]
warnings: list[str]
valid: bool
class RequirementResponse(BaseModel):
id: str
title: str
description: str | None
priority: Priority
phase: int | None
status: Status
node_type: NodeType
parent_id: str | None
sort_order: int
project_id: int | None
gitea_issue_number: int | None
gitea_repo: str | None
traceable: bool
measurable: bool
consistent: bool
single_purpose: bool
acceptance_criteria: str | None
test_spec: str | None
test_code: str | None
impl_code: str | None
code_language: CodeLanguage | None
target_test_file: str | None
target_impl_file: str | None
test_status: TestStatus
created_at: datetime
updated_at: datetime
quality: QualityReport | None = None
children: list["RequirementResponse"] = []
model_config = {"from_attributes": True}
class RequirementTreeNode(BaseModel):
id: str
title: str
priority: Priority
phase: int | None
status: Status
node_type: NodeType
parent_id: str | None
child_count: int = 0
traceable: bool
measurable: bool
consistent: bool
single_purpose: bool
gitea_issue_number: int | None = None
children: list["RequirementTreeNode"] = []
model_config = {"from_attributes": True}
# ─── Requirement History ────────────────────────────────────────────────
class RequirementHistorySummary(BaseModel):
id: int
requirement_id: str
version: int
title: str
status: str
created_at: datetime
change_summary: str | None
model_config = {"from_attributes": True}
class RequirementHistoryDetail(RequirementHistorySummary):
description: str | None
priority: str
phase: int | None
node_type: str
acceptance_criteria: str | None
test_spec: str | None
test_code: str | None
impl_code: str | None
code_language: str | None
target_test_file: str | None
target_impl_file: str | None
measurable: bool
consistent: bool
single_purpose: bool
gitea_issue_number: int | None
gitea_repo: str | None
# ─── Graph Layouts ──────────────────────────────────────────────────────
class GraphLayoutCreate(BaseModel):
name: str = Field(..., max_length=128)
description: str | None = None
project_id: int | None = None
node_positions: dict[str, dict[str, float]]
viewport: dict[str, object]
is_default: bool = False
class GraphLayoutUpdate(BaseModel):
name: str | None = None
description: str | None = None
node_positions: dict[str, dict[str, float]] | None = None
viewport: dict[str, object] | None = None
is_default: bool | None = None
class GraphLayoutResponse(BaseModel):
id: int
name: str
description: str | None
project_id: int | None
is_default: bool
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class GraphLayoutDetailResponse(GraphLayoutResponse):
node_positions: dict[str, dict[str, float]]
viewport: dict[str, object]
# ─── Cross-Pillar Links ─────────────────────────────────────────────────
class LinkCreate(BaseModel):
source_id: str
target_id: str
relationship_type: str = "shared_parent"
class LinkResponse(BaseModel):
source_id: str
target_id: str
relationship_type: str
model_config = {"from_attributes": True}
# ─── Project ────────────────────────────────────────────────────────────
class ProjectCreate(BaseModel):
name: str = Field(..., max_length=256)
description: str | None = None
repo_path: str | None = None
repo_url: str | None = None
requirements_yaml_path: str | None = None
class ProjectResponse(BaseModel):
id: int
name: str
description: str | None
repo_path: str | None
repo_url: str | None
requirements_yaml_path: str | None
created_at: datetime
model_config = {"from_attributes": True}
# ─── Stats ───────────────────────────────────────────────────────────────
class StatsResponse(BaseModel):
total_nodes: int
pillars: int
requirements: int
sub_requirements: int
leaves: int
by_status: dict[str, int]
by_priority: dict[str, int]
quality_passing: int
quality_failing: int
coverage_percent: float