Lars
d1bc725ab5
Some checks failed
Deploy Development / deploy (push) Failing after 48s
Test Suite / pytest-backend (push) Failing after 1s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 3s
Test Suite / compose-smoke (push) Has been skipped
RoadmapItem type work_cycle, Sprint-API, agile_iteration Next-Action-Strategie; Steering-Snapshot zeigt active_work_cycle. Co-authored-by: Cursor <cursoragent@cursor.com>
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
"""Work cycle (Sprint/Zeitbox) API — AP2.0f."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from typing import Literal, Optional
|
|
|
|
from capabilities import require_capability
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from services import work_cycle as work_cycle_service
|
|
from tenant_context import TenantContext
|
|
|
|
initiative_router = APIRouter(
|
|
prefix="/api/initiatives/{initiative_id}/work-cycles",
|
|
tags=["work-cycles"],
|
|
)
|
|
|
|
|
|
class WorkCycleCreateRequest(BaseModel):
|
|
title: str = Field(min_length=1, max_length=255)
|
|
goal_description: str = ""
|
|
status: Literal["planned", "active"] = "planned"
|
|
target_date: Optional[date] = None
|
|
sort_order: int = 0
|
|
|
|
|
|
@initiative_router.get("")
|
|
def list_work_cycles(
|
|
initiative_id: str,
|
|
ctx: TenantContext = Depends(require_capability("kairo.milestone.read")),
|
|
):
|
|
return work_cycle_service.list_work_cycles_for_initiative(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
|
|
|
|
@initiative_router.get("/active")
|
|
def get_active_work_cycle(
|
|
initiative_id: str,
|
|
ctx: TenantContext = Depends(require_capability("kairo.milestone.read")),
|
|
):
|
|
item = work_cycle_service.get_active_work_cycle(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
if not item:
|
|
return None
|
|
item["open_action_count"] = work_cycle_service.count_actions_in_work_cycle(
|
|
tenant_id=ctx.tenant_id,
|
|
initiative_id=initiative_id,
|
|
work_cycle_id=item["id"],
|
|
)
|
|
return item
|
|
|
|
|
|
@initiative_router.post("", status_code=201)
|
|
def create_work_cycle(
|
|
initiative_id: str,
|
|
body: WorkCycleCreateRequest,
|
|
ctx: TenantContext = Depends(require_capability("kairo.milestone.manage")),
|
|
):
|
|
try:
|
|
item = work_cycle_service.create_work_cycle(
|
|
tenant_id=ctx.tenant_id,
|
|
initiative_id=initiative_id,
|
|
title=body.title,
|
|
goal_description=body.goal_description,
|
|
status=body.status,
|
|
target_date=body.target_date,
|
|
sort_order=body.sort_order,
|
|
user_id=ctx.user_id,
|
|
)
|
|
if body.status == "active":
|
|
item = work_cycle_service.activate_work_cycle(
|
|
tenant_id=ctx.tenant_id,
|
|
initiative_id=initiative_id,
|
|
work_cycle_id=item["id"],
|
|
user_id=ctx.user_id,
|
|
)
|
|
return item
|
|
except ValueError as exc:
|
|
detail = str(exc)
|
|
if detail == "Initiative nicht gefunden":
|
|
raise HTTPException(status_code=404, detail=detail) from exc
|
|
raise HTTPException(status_code=400, detail=detail) from exc
|
|
|
|
|
|
@initiative_router.post("/{work_cycle_id}/activate")
|
|
def activate_work_cycle(
|
|
initiative_id: str,
|
|
work_cycle_id: str,
|
|
ctx: TenantContext = Depends(require_capability("kairo.milestone.manage")),
|
|
):
|
|
try:
|
|
return work_cycle_service.activate_work_cycle(
|
|
tenant_id=ctx.tenant_id,
|
|
initiative_id=initiative_id,
|
|
work_cycle_id=work_cycle_id,
|
|
user_id=ctx.user_id,
|
|
)
|
|
except ValueError as exc:
|
|
detail = str(exc)
|
|
if detail in ("Initiative nicht gefunden", "Zeitbox nicht gefunden"):
|
|
raise HTTPException(status_code=404, detail=detail) from exc
|
|
raise HTTPException(status_code=400, detail=detail) from exc
|