from kodosumi.core import ServeAPI, Launch, Tracer
from kodosumi.service.inputs.forms import Model, InputText, Checkbox, Submit, Cancel, Markdown
from kodosumi.service.inputs.errors import InputsError
from fastapi import Request
app = ServeAPI()
# Entry point form
entry_form = Model(
InputText(label="Document Name", name="doc_name", placeholder="Enter document name"),
InputText(label="Author", name="author", placeholder="Enter author name"),
Submit("Start Review"),
Cancel("Cancel"),
)
@app.enter(
"/review",
model=entry_form,
summary="Document Review Workflow",
description="Review workflow with human approval steps",
)
async def start_review(inputs: dict, request: Request) -> Launch:
return Launch(request, "document_service:review_runner", inputs=inputs)
# Runner function
async def review_runner(inputs: dict, tracer: Tracer):
await tracer.debug(f"Starting review for document: {inputs['doc_name']}")
# First approval - content review
content_approval = await tracer.lock("content-review", data={
"document": inputs['doc_name'],
"author": inputs['author'],
"stage": "content"
})
# Second approval - final review
final_approval = await tracer.lock("final-review", data={
"document": inputs['doc_name'],
"content_approved": content_approval,
"stage": "final"
})
return {
"document": inputs['doc_name'],
"content_approval": content_approval,
"final_approval": final_approval,
"status": "completed"
}
# Content review lock
@app.lock("content-review")
async def content_review_lock(data: dict):
return Model(
Markdown(f"# Content Review Required\n\n**Document**: {data['document']}\n**Author**: {data['author']}"),
Checkbox(label="Content is accurate", name="accurate", option="ACCURATE", value=False),
Checkbox(label="Grammar is correct", name="grammar", option="GRAMMAR", value=False),
InputText(label="Review Comments", name="comments", placeholder="Enter review comments"),
Submit("Approve Content"),
Cancel("Request Changes"),
)
# Content review lease
@app.lease("content-review")
async def content_review_lease(inputs: dict):
if not inputs.get("accurate") or not inputs.get("grammar"):
raise InputsError(
accurate="Content accuracy must be confirmed",
grammar="Grammar must be verified"
)
return {
"approved": True,
"comments": inputs.get("comments", ""),
"reviewer": "human_reviewer"
}
# Final review lock
@app.lock("final-review")
async def final_review_lock(data: dict):
return Model(
Markdown(f"# Final Review\n\n**Document**: {data['document']}\n**Content Approved**: {data['content_approved']['approved']}"),
Checkbox(label="Ready for publication", name="publish", option="PUBLISH", value=False),
InputText(label="Final Comments", name="final_comments", placeholder="Final approval comments"),
Submit("Publish"),
Cancel("Hold"),
)
# Final review lease
@app.lease("final-review")
async def final_review_lease(inputs: dict):
if not inputs.get("publish"):
raise InputsError(publish="Publication approval is required")
return {
"published": True,
"comments": inputs.get("final_comments", ""),
"publisher": "human_publisher"
}