feat: hand off assigned pull requests (Closes #567)

src/gitea_proxy.py

@@ -1498,6 +1498,92 @@ async def handoff_assigned_issue(
     }
 
 
+async def pull_handoff_candidates(repository: str) -> list[dict]:
+    candidates = await issue_handoff_candidates(repository)
+    return [
+        item for item in candidates
+        if re.fullmatch(r"[A-Za-z0-9_.-]+", item["login"])
+    ][:25]
+
+
+async def _change_assigned_pull_owners(
+    repository: str, number: int, recipient: str | None
+) -> dict:
+    login, pull = await _current_login_and_target(
+        f"repos/{repository}/pulls/{number}"
+    )
+    assignees_value = pull.get("assignees")
+    assignees = assignees_value if isinstance(assignees_value, list) else []
+    login_folded = login.casefold()
+    assigned_to_login = any(
+        isinstance(item, dict)
+        and isinstance(item.get("login"), str)
+        and item["login"].casefold() == login_folded
+        for item in assignees
+    )
+    if (
+        pull.get("state") != "open"
+        or pull.get("merged") is True
+        or not assigned_to_login
+    ):
+        raise IssueNotAvailableError("Pull request is not assigned to the current user")
+
+    if recipient is not None:
+        eligible = {
+            item["login"] for item in await pull_handoff_candidates(repository)
+        }
+        if recipient not in eligible:
+            raise IssueNotAvailableError("Handoff recipient is not eligible")
+
+    desired = [
+        item["login"] for item in assignees
+        if isinstance(item, dict)
+        and isinstance(item.get("login"), str)
+        and item["login"].casefold() != login_folded
+    ]
+    if recipient is not None and recipient not in desired:
+        desired.append(recipient)
+    response = await _get_client().patch(
+        f"/api/v1/repos/{repository}/issues/{number}",
+        headers=_auth(),
+        json={"assignees": desired},
+    )
+    response.raise_for_status()
+    confirmed = response.json()
+    confirmed_value = confirmed.get("assignees") if isinstance(confirmed, dict) else None
+    confirmed_assignees = [
+        item["login"] for item in confirmed_value
+        if isinstance(item, dict) and isinstance(item.get("login"), str)
+    ] if isinstance(confirmed_value, list) else []
+    if (
+        not isinstance(confirmed, dict)
+        or confirmed.get("number") != number
+        or login_folded in {item.casefold() for item in confirmed_assignees}
+        or set(confirmed_assignees) != set(desired)
+        or (recipient is not None and recipient not in confirmed_assignees)
+    ):
+        raise ValueError("Gitea did not confirm pull request ownership change")
+    result = {
+        "repository": repository,
+        "number": number,
+        "state": confirmed.get("state", "open"),
+        "assignees": confirmed_assignees,
+    }
+    if recipient is not None:
+        result["recipient"] = recipient
+    return result
+
+
+async def handoff_assigned_pull(
+    repository: str, number: int, recipient: str
+) -> dict:
+    return await _change_assigned_pull_owners(repository, number, recipient)
+
+
+async def release_assigned_pull(repository: str, number: int) -> dict:
+    return await _change_assigned_pull_owners(repository, number, None)
+
+
 async def update_issue_labels(repository: str, number: int, label_ids: list[int]) -> dict:
     response = await _get_client().patch(
         f"/api/v1/repos/{repository}/issues/{number}",

src/main.py

@@ -4317,6 +4317,77 @@ async def assigned_pull_checks(
         )
 
 
+@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/handoff-candidates")
+async def pull_handoff_candidates(
+    owner: str, repo: str, number: int = PathParam(gt=0)
+) -> JSONResponse:
+    repository = f"{owner}/{repo}"
+    try:
+        if not await gitea_proxy.is_assigned_pull(repository, number):
+            raise HTTPException(status_code=404, detail="Assigned pull request not found")
+        result = await asyncio.wait_for(
+            gitea_proxy.pull_handoff_candidates(repository),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
+    except HTTPException:
+        raise
+    except Exception:
+        return JSONResponse(
+            {"error": "Teammates could not be loaded. Please retry."},
+            status_code=503,
+            headers={"Retry-After": "1"},
+        )
+    return JSONResponse(result, headers={"Cache-Control": "no-store"})
+
+
+@app.patch("/api/v1/repos/{owner}/{repo}/pulls/{number}/handoff")
+async def handoff_assigned_pull(
+    handoff: IssueHandoff,
+    owner: str,
+    repo: str,
+    number: int = PathParam(gt=0),
+) -> JSONResponse:
+    try:
+        result = await asyncio.wait_for(
+            gitea_proxy.handoff_assigned_pull(
+                f"{owner}/{repo}", number, handoff.recipient
+            ),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
+    except gitea_proxy.IssueNotAvailableError:
+        return JSONResponse(
+            {"error": "The pull request or recipient changed. Reload before handing off."},
+            status_code=409,
+        )
+    except Exception:
+        return JSONResponse(
+            {"error": "The handoff could not be confirmed. It remains in My Work; please retry."},
+            status_code=503,
+            headers={"Retry-After": "1"},
+        )
+    return JSONResponse(result)
+
+
+@app.patch("/api/v1/repos/{owner}/{repo}/pulls/{number}/release")
+async def release_assigned_pull(
+    owner: str, repo: str, number: int = PathParam(gt=0)
+) -> JSONResponse:
+    try:
+        result = await asyncio.wait_for(
+            gitea_proxy.release_assigned_pull(f"{owner}/{repo}", number),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
+    except gitea_proxy.IssueNotAvailableError:
+        raise HTTPException(status_code=404, detail="Assigned pull request not found")
+    except Exception:
+        return JSONResponse(
+            {"error": "The assignment could not be released. It remains in My Work; please retry."},
+            status_code=503,
+            headers={"Retry-After": "1"},
+        )
+    return JSONResponse(result)
+
+
 @app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments")
 async def assigned_pull_conversation(
     owner: str,