1781 lines
64 KiB
Python
1781 lines
64 KiB
Python
import asyncio
|
|
import math
|
|
import os
|
|
import time
|
|
from collections.abc import Awaitable, Coroutine
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from fastapi import FastAPI, Header, HTTPException, Path as PathParam, Query
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field, PositiveInt, field_validator, model_validator
|
|
|
|
from src import gitea_proxy
|
|
from src.gitea_proxy import (
|
|
activity_events,
|
|
current_user,
|
|
is_requested_review,
|
|
issues,
|
|
mark_notification_read,
|
|
notification_detail,
|
|
notifications,
|
|
pull_requests,
|
|
pull_review_detail,
|
|
repos,
|
|
)
|
|
from src.idempotency import IdempotencyLedger
|
|
from src.models import Issue, Milestone, PullRequest, Repo, User
|
|
from src.suggestion_engine import compute
|
|
from src.views import router as frontend_router
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI):
|
|
global _live_snapshot_task, _available_issue_snapshot_task
|
|
gitea_proxy.start_client()
|
|
try:
|
|
yield
|
|
finally:
|
|
live_task = _live_snapshot_task
|
|
available_task = _available_issue_snapshot_task
|
|
for task in (live_task, available_task):
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
try:
|
|
await gitea_proxy.stop_client()
|
|
finally:
|
|
if _live_snapshot_task is live_task:
|
|
_live_snapshot_task = None
|
|
if _available_issue_snapshot_task is available_task:
|
|
_available_issue_snapshot_task = None
|
|
|
|
|
|
app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
|
|
CONTEXT_TIMEOUT_SECONDS = 5.0
|
|
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
|
|
READINESS_TIMEOUT_SECONDS = 5.0
|
|
REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
|
|
ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
|
|
ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
|
|
ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES = 256
|
|
AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS = 600.0
|
|
AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES = 256
|
|
NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
|
|
NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
|
|
WORK_PAGE_TIMEOUT_SECONDS = 5.0
|
|
GLOBAL_SEARCH_TIMEOUT_SECONDS = 3.0
|
|
NOTIFICATION_DETAIL_TIMEOUT_SECONDS = 5.0
|
|
BULK_NOTIFICATION_CONCURRENCY = 5
|
|
BULK_NOTIFICATION_DEADLINE_SECONDS = 6.0
|
|
LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
|
|
LIVE_SNAPSHOT_RETRY_BASE_SECONDS = 5.0
|
|
LIVE_SNAPSHOT_RETRY_MAX_SECONDS = 60.0
|
|
AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS = 15.0
|
|
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
|
_live_snapshot_task: asyncio.Task | None = None
|
|
_live_snapshot_value: dict | None = None
|
|
_live_snapshot_created_at: float | None = None
|
|
LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
|
|
_live_section_created_at: dict[str, float | None] = {
|
|
section: None for section in LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
_live_section_failure_count: dict[str, int] = {
|
|
section: 0 for section in LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
_live_section_retry_at: dict[str, float | None] = {
|
|
section: None for section in LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
_live_snapshot_refreshing_sections: set[str] = set()
|
|
_read_notification_ids: set[int] = set()
|
|
_authored_action_operations: dict[
|
|
str, tuple[tuple[Any, ...], asyncio.Task, float]
|
|
] = {}
|
|
_state_dir = Path(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state"))
|
|
_idempotency_ledger = IdempotencyLedger(
|
|
os.getenv("STACKCHAIN_IDEMPOTENCY_DB", str(_state_dir / "idempotency.sqlite3")),
|
|
ttl_seconds=AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS,
|
|
max_entries=(
|
|
AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES
|
|
+ ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES
|
|
),
|
|
)
|
|
_available_issue_snapshot_task: asyncio.Task | None = None
|
|
_available_issue_snapshot_value: list[dict] | None = None
|
|
_available_issue_snapshot_created_at: float | None = None
|
|
|
|
|
|
class ContextPayloadError(ValueError):
|
|
"""Raised when Gitea returns a structurally invalid context payload."""
|
|
|
|
|
|
class ReadinessPayloadError(ValueError):
|
|
"""Raised when Gitea returns a structurally invalid readiness payload."""
|
|
|
|
|
|
class NotificationReadBatch(BaseModel):
|
|
ids: list[PositiveInt] = Field(min_length=1, max_length=50)
|
|
|
|
|
|
class NotificationReply(BaseModel):
|
|
body: str = Field(min_length=1, max_length=10_000)
|
|
|
|
@field_validator("body")
|
|
@classmethod
|
|
def strip_body(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("reply must not be blank")
|
|
return value
|
|
|
|
|
|
class IssueComment(BaseModel):
|
|
body: str = Field(min_length=1, max_length=10_000)
|
|
|
|
@field_validator("body")
|
|
@classmethod
|
|
def strip_body(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("comment must not be blank")
|
|
return value
|
|
|
|
|
|
class IssueCreation(BaseModel):
|
|
title: str = Field(min_length=1, max_length=255)
|
|
body: str = Field(default="", max_length=10_000)
|
|
label_ids: list[int] = Field(default_factory=list, max_length=20)
|
|
|
|
@field_validator("title")
|
|
@classmethod
|
|
def strip_title(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("title must not be blank")
|
|
return value
|
|
|
|
@field_validator("body")
|
|
@classmethod
|
|
def strip_issue_body(cls, value: str) -> str:
|
|
return value.strip()
|
|
|
|
|
|
class IssueContentUpdate(BaseModel):
|
|
title: str = Field(min_length=1, max_length=255)
|
|
body: str = Field(default="", max_length=10_000)
|
|
expected_updated_at: str = Field(min_length=1, max_length=64)
|
|
|
|
@field_validator("title")
|
|
@classmethod
|
|
def strip_title(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("title must not be blank")
|
|
return value
|
|
|
|
@field_validator("body")
|
|
@classmethod
|
|
def strip_body(cls, value: str) -> str:
|
|
return value.strip()
|
|
|
|
|
|
class IssueLabelUpdate(BaseModel):
|
|
label_ids: list[PositiveInt] = Field(max_length=20)
|
|
|
|
|
|
class IssueDueDateUpdate(BaseModel):
|
|
due_date: str | None = Field(
|
|
default=None,
|
|
pattern=r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$",
|
|
max_length=20,
|
|
)
|
|
|
|
|
|
class IssueMilestoneUpdate(BaseModel):
|
|
milestone_id: PositiveInt | None = None
|
|
|
|
|
|
class PullReviewComment(BaseModel):
|
|
path: str = Field(min_length=1, max_length=1_000)
|
|
body: str = Field(min_length=1, max_length=10_000)
|
|
new_position: PositiveInt | None = None
|
|
old_position: PositiveInt | None = None
|
|
|
|
@field_validator("path", "body")
|
|
@classmethod
|
|
def strip_comment_text(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("inline comment values must not be blank")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def require_one_position(self):
|
|
if (self.new_position is None) == (self.old_position is None):
|
|
raise ValueError("inline comments require exactly one old or new position")
|
|
return self
|
|
|
|
|
|
class PullReviewSubmission(BaseModel):
|
|
decision: str
|
|
body: str = Field(max_length=10_000)
|
|
expected_head_sha: str
|
|
comments: list[PullReviewComment] = Field(default_factory=list, max_length=50)
|
|
|
|
@field_validator("decision")
|
|
@classmethod
|
|
def validate_decision(cls, value: str) -> str:
|
|
if value not in {"comment", "approve", "request_changes"}:
|
|
raise ValueError("unsupported review decision")
|
|
return value
|
|
|
|
|
|
class PullMergeSubmission(BaseModel):
|
|
expected_head_sha: str = Field(min_length=1, max_length=128)
|
|
|
|
|
|
async def _run_idempotent_authored_action(
|
|
operation: Coroutine[Any, Any, Any],
|
|
*,
|
|
idempotency_key: str | None,
|
|
fingerprint: tuple[Any, ...],
|
|
timeout: float,
|
|
):
|
|
if not idempotency_key:
|
|
return await asyncio.wait_for(operation, timeout=timeout)
|
|
|
|
reservation = _idempotency_ledger.reserve(idempotency_key, fingerprint)
|
|
if reservation.state == "conflict":
|
|
operation.close()
|
|
raise HTTPException(status_code=409, detail="Idempotency key already used")
|
|
if reservation.state == "completed":
|
|
operation.close()
|
|
return reservation.response
|
|
if reservation.state == "busy":
|
|
operation.close()
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="Authored action queue is busy; please retry",
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
existing = _authored_action_operations.get(idempotency_key)
|
|
if reservation.state == "pending":
|
|
operation.close()
|
|
if existing is None or existing[0] != fingerprint:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail=(
|
|
"This action may still be completing; verify its result before retrying"
|
|
),
|
|
headers={"Retry-After": "5"},
|
|
)
|
|
task = existing[1]
|
|
else:
|
|
async def persist_result():
|
|
result = await operation
|
|
_idempotency_ledger.complete(idempotency_key, result)
|
|
return result
|
|
|
|
task = asyncio.create_task(persist_result())
|
|
_authored_action_operations[idempotency_key] = (
|
|
fingerprint,
|
|
task,
|
|
time.monotonic(),
|
|
)
|
|
|
|
def discard_completed(completed: asyncio.Task) -> None:
|
|
existing_operation = _authored_action_operations.get(idempotency_key)
|
|
if existing_operation is not None and existing_operation[1] is completed:
|
|
_authored_action_operations.pop(idempotency_key, None)
|
|
|
|
task.add_done_callback(discard_completed)
|
|
return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
|
|
|
|
|
|
def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict:
|
|
user_model = User(
|
|
id=user_data["id"],
|
|
login=user_data["login"],
|
|
full_name=user_data.get("full_name") or "",
|
|
email=user_data.get("email") or "",
|
|
)
|
|
repo_models = [
|
|
Repo(
|
|
id=r["id"], name=r["name"], full_name=r["full_name"],
|
|
description=r.get("description") or "", url=r["html_url"],
|
|
updated_at=r.get("updated_at", ""),
|
|
)
|
|
for r in (repo_data or [])[:50]
|
|
if isinstance(r, dict)
|
|
and all(field in r for field in ("id", "name", "full_name", "html_url"))
|
|
]
|
|
issue_models = [
|
|
Issue(
|
|
id=i["id"], number=i["number"], title=i["title"], state=i["state"],
|
|
labels=[label.get("name", "") for label in (i.get("labels") or []) if isinstance(label, dict)],
|
|
assignees=[assignee.get("login", "") for assignee in (i.get("assignees") or []) if isinstance(assignee, dict)],
|
|
repository=i["repository"].get("full_name", "") if isinstance(i.get("repository"), dict) else "",
|
|
updated_at=i.get("updated_at") or "",
|
|
due_date=i.get("due_date") if isinstance(i.get("due_date"), str) else None,
|
|
milestone=Milestone(**i["milestone"]) if isinstance(i.get("milestone"), dict)
|
|
and isinstance(i["milestone"].get("id"), int)
|
|
and isinstance(i["milestone"].get("title"), str) else None,
|
|
url=i["html_url"],
|
|
)
|
|
for i in (issues_data or [])[:50]
|
|
if isinstance(i, dict)
|
|
and all(field in i for field in ("id", "number", "title", "state", "html_url"))
|
|
]
|
|
pr_models = [
|
|
PullRequest(
|
|
id=p["id"], number=p["number"], title=p["title"], state=p["state"],
|
|
user=p["user"].get("login", "") if isinstance(p.get("user"), dict) else "",
|
|
labels=[label.get("name", "") for label in (p.get("labels") or []) if isinstance(label, dict)],
|
|
assignees=[assignee.get("login", "") for assignee in (p.get("assignees") or []) if isinstance(assignee, dict)],
|
|
work_reasons=[reason for reason in (p.get("work_reasons") or []) if reason in ("assigned_to_me", "review_requested")],
|
|
repository=p["repository"].get("full_name", "") if isinstance(p.get("repository"), dict) else "",
|
|
updated_at=p.get("updated_at") or "", url=p["html_url"],
|
|
)
|
|
for p in (prs_data or [])[:50]
|
|
if isinstance(p, dict)
|
|
and all(field in p for field in ("id", "number", "title", "state", "html_url"))
|
|
]
|
|
payload = compute(user_model, repo_models, issue_models, pr_models).model_dump()
|
|
pagination = {}
|
|
pagination.update(getattr(issues_data, "pagination", {}))
|
|
pagination.update(getattr(prs_data, "pagination", {}))
|
|
if pagination:
|
|
payload["work_pagination"] = pagination
|
|
return payload
|
|
|
|
|
|
def _normalize_work_items(stream: str, items: list[dict]) -> list[dict]:
|
|
if stream == "issue":
|
|
return [
|
|
Issue(
|
|
id=item["id"], number=item["number"], title=item["title"],
|
|
state=item["state"],
|
|
labels=[label.get("name", "") for label in (item.get("labels") or []) if isinstance(label, dict)],
|
|
assignees=[assignee.get("login", "") for assignee in (item.get("assignees") or []) if isinstance(assignee, dict)],
|
|
repository=item["repository"].get("full_name", "") if isinstance(item.get("repository"), dict) else "",
|
|
updated_at=item.get("updated_at") or "",
|
|
due_date=item.get("due_date") if isinstance(item.get("due_date"), str) else None,
|
|
milestone=Milestone(**item["milestone"])
|
|
if isinstance(item.get("milestone"), dict)
|
|
and isinstance(item["milestone"].get("id"), int)
|
|
and isinstance(item["milestone"].get("title"), str) else None,
|
|
url=item["html_url"],
|
|
).model_dump()
|
|
for item in items
|
|
if all(field in item for field in ("id", "number", "title", "state", "html_url"))
|
|
]
|
|
reason = "assigned_to_me" if stream == "pull" else "review_requested"
|
|
return [
|
|
PullRequest(
|
|
id=item["id"], number=item["number"], title=item["title"],
|
|
state=item["state"],
|
|
user=item["user"].get("login", "") if isinstance(item.get("user"), dict) else "",
|
|
labels=[label.get("name", "") for label in (item.get("labels") or []) if isinstance(label, dict)],
|
|
assignees=[assignee.get("login", "") for assignee in (item.get("assignees") or []) if isinstance(assignee, dict)],
|
|
work_reasons=[reason],
|
|
repository=item["repository"].get("full_name", "") if isinstance(item.get("repository"), dict) else "",
|
|
updated_at=item.get("updated_at") or "", url=item["html_url"],
|
|
).model_dump()
|
|
for item in items
|
|
if all(field in item for field in ("id", "number", "title", "state", "html_url"))
|
|
]
|
|
|
|
app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
|
|
app.include_router(frontend_router)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def prevent_live_api_caching(request, call_next):
|
|
response = await call_next(request)
|
|
if request.url.path in {"/api/v1/context", "/api/v1/events", "/api/v1/live", "/api/v1/available-issues", "/api/v1/search", "/api/v1/work-route"} or request.url.path.startswith("/api/v1/work/") or (
|
|
request.url.path.startswith("/api/v1/repos/")
|
|
and request.url.path.endswith("/review")
|
|
) or request.url.path.startswith("/api/v1/notifications") or (
|
|
request.url.path.startswith("/api/v1/repos/")
|
|
and ("/issues/" in request.url.path or "/pulls/" in request.url.path)
|
|
) or (
|
|
request.url.path.startswith("/api/v1/repos/")
|
|
and request.url.path.endswith("/issues")
|
|
) or (
|
|
request.url.path.startswith("/api/v1/repos/")
|
|
and request.url.path.endswith(("/labels", "/milestones"))
|
|
):
|
|
response.headers["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
@app.get("/healthz")
|
|
def health() -> dict[str, str]:
|
|
"""Return process liveness without depending on Gitea."""
|
|
return {"status": "ok", "service": "stackchain-dashboard"}
|
|
|
|
|
|
@app.get("/readyz")
|
|
async def readiness():
|
|
"""Return readiness after verifying the configured Gitea connection."""
|
|
try:
|
|
user = await asyncio.wait_for(
|
|
current_user(), timeout=READINESS_TIMEOUT_SECONDS
|
|
)
|
|
if not isinstance(user, dict) or not user.get("login"):
|
|
raise ReadinessPayloadError(
|
|
"Gitea current-user response did not include a login"
|
|
)
|
|
except Exception as exc:
|
|
timed_out = isinstance(exc, TimeoutError)
|
|
error_message = (
|
|
f"Gitea readiness check timed out after {READINESS_TIMEOUT_SECONDS:g}s"
|
|
if timed_out
|
|
else (
|
|
str(exc)
|
|
if isinstance(exc, ReadinessPayloadError)
|
|
else "Gitea readiness check is temporarily unavailable"
|
|
)
|
|
)
|
|
return JSONResponse(
|
|
{
|
|
"status": "not_ready",
|
|
"service": "stackchain-dashboard",
|
|
"error": error_message,
|
|
},
|
|
status_code=503,
|
|
headers={
|
|
"Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
|
|
} if timed_out else None,
|
|
)
|
|
return {
|
|
"status": "ready",
|
|
"service": "stackchain-dashboard",
|
|
"gitea_user": user["login"],
|
|
}
|
|
|
|
|
|
@app.get("/api/v1/context")
|
|
async def context() -> JSONResponse:
|
|
try:
|
|
user_data, repo_data, issues_data, prs_data = await asyncio.wait_for(
|
|
asyncio.gather(current_user(), repos(), issues(), pull_requests()),
|
|
timeout=CONTEXT_TIMEOUT_SECONDS,
|
|
)
|
|
if not isinstance(user_data, dict):
|
|
raise ContextPayloadError("Gitea current-user response was not an object")
|
|
if not all(field in user_data for field in ("id", "login")):
|
|
raise ContextPayloadError(
|
|
"Gitea current-user response did not include id and login"
|
|
)
|
|
if any(
|
|
data is not None and not isinstance(data, list)
|
|
for data in (repo_data, issues_data, prs_data)
|
|
):
|
|
raise ContextPayloadError("Gitea collection response was not a list")
|
|
except Exception as e:
|
|
if isinstance(e, TimeoutError):
|
|
error_message = (
|
|
f"Gitea context request timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"
|
|
)
|
|
elif isinstance(e, ContextPayloadError):
|
|
error_message = str(e)
|
|
else:
|
|
error_message = "Gitea context is temporarily unavailable"
|
|
return JSONResponse({
|
|
"user": {"id": None, "login": "timmy", "full_name": "Timmy Jr", "email": ""},
|
|
"repos": [],
|
|
"issues": [],
|
|
"pull_requests": [],
|
|
"view": "dashboard",
|
|
"deltas": [
|
|
{"panel": "auth", "action": "connect", "target": "stackchain-dashboard backend", "priority": "high"},
|
|
{"panel": "gitea", "action": "set GITEA_URL + GITEA_TOKEN", "target": "/root/stackchain-dashboard/.env or systemd env", "priority": "high"},
|
|
],
|
|
"error": error_message,
|
|
}, headers={
|
|
"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))
|
|
} if isinstance(e, TimeoutError) else None)
|
|
return JSONResponse(_context_payload(user_data, repo_data, issues_data, prs_data))
|
|
|
|
|
|
@app.get("/api/v1/search")
|
|
async def global_search(
|
|
q: str = Query(min_length=2, max_length=100),
|
|
limit: int = Query(default=10, ge=1, le=25),
|
|
) -> JSONResponse:
|
|
query = q.strip()
|
|
if len(query) < 2:
|
|
raise HTTPException(status_code=422, detail="Search query must contain at least 2 characters")
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.global_search(query, limit),
|
|
timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Search is temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse({"query": query, **result})
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/preview")
|
|
async def global_search_preview(
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
kind: Literal["issue", "pull"] = Query(),
|
|
) -> JSONResponse:
|
|
try:
|
|
preview = await asyncio.wait_for(
|
|
gitea_proxy.work_preview(f"{owner}/{repo}", kind, number),
|
|
timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "This work item is temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(preview)
|
|
|
|
|
|
@app.get("/api/v1/work-route")
|
|
async def resolve_work_route(
|
|
kind: Literal["issue", "pull", "review", "update"] = Query(),
|
|
repository: str | None = Query(default=None, pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"),
|
|
number: int | None = Query(default=None, gt=0),
|
|
notification_id: int | None = Query(default=None, gt=0),
|
|
) -> JSONResponse:
|
|
repository_route = repository is not None and number is not None and notification_id is None
|
|
update_route = kind == "update" and notification_id is not None and repository is None and number is None
|
|
if (kind == "update" and not update_route) or (kind != "update" and not repository_route):
|
|
raise HTTPException(status_code=422, detail="Invalid work route identity")
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.resolve_work_route(kind, repository, number, notification_id),
|
|
timeout=WORK_PAGE_TIMEOUT_SECONDS,
|
|
)
|
|
except gitea_proxy.WorkRouteUnavailableError:
|
|
raise HTTPException(status_code=404, detail="Work item is no longer in My Work")
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The shared work item is temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.get("/api/v1/work/{stream}")
|
|
async def paged_work(
|
|
stream: Literal["issue", "pull", "review"],
|
|
page: int = Query(ge=2, le=100),
|
|
) -> JSONResponse:
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.work_page(stream, page), timeout=WORK_PAGE_TIMEOUT_SECONDS
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Work page is temporarily unavailable"},
|
|
status_code=503,
|
|
headers={"Retry-After": str(math.ceil(WORK_PAGE_TIMEOUT_SECONDS))},
|
|
)
|
|
return JSONResponse({
|
|
"stream": stream,
|
|
"page": result["page"],
|
|
"total": result["total"],
|
|
"has_more": result["has_more"],
|
|
"items": _normalize_work_items(stream, result["items"]),
|
|
})
|
|
|
|
|
|
async def _refresh_available_issue_snapshot() -> list[dict]:
|
|
global _available_issue_snapshot_value, _available_issue_snapshot_created_at
|
|
result = await gitea_proxy.available_issue_snapshot()
|
|
_available_issue_snapshot_value = result
|
|
_available_issue_snapshot_created_at = time.monotonic()
|
|
return result
|
|
|
|
|
|
async def _available_issue_snapshot() -> tuple[list[dict], bool]:
|
|
global _available_issue_snapshot_task
|
|
now = time.monotonic()
|
|
if (
|
|
_available_issue_snapshot_value is not None
|
|
and _available_issue_snapshot_created_at is not None
|
|
and now - _available_issue_snapshot_created_at
|
|
< AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS
|
|
):
|
|
return _available_issue_snapshot_value, False
|
|
if _available_issue_snapshot_task is None or _available_issue_snapshot_task.done():
|
|
_available_issue_snapshot_task = asyncio.create_task(
|
|
_refresh_available_issue_snapshot()
|
|
)
|
|
try:
|
|
return await asyncio.shield(_available_issue_snapshot_task), False
|
|
except Exception:
|
|
if _available_issue_snapshot_value is not None:
|
|
return _available_issue_snapshot_value, True
|
|
raise
|
|
|
|
|
|
def _available_issue_page(items: list[dict], page: int, limit: int = 50) -> dict:
|
|
start = (page - 1) * limit
|
|
page_items = items[start:start + limit]
|
|
return {
|
|
"items": page_items,
|
|
"page": page,
|
|
"total": len(items),
|
|
"has_more": start + len(page_items) < len(items),
|
|
}
|
|
|
|
|
|
@app.get("/api/v1/available-issues")
|
|
async def available_issues(page: int = Query(default=1, ge=1, le=100)) -> JSONResponse:
|
|
try:
|
|
items, stale = await asyncio.wait_for(
|
|
_available_issue_snapshot(), timeout=WORK_PAGE_TIMEOUT_SECONDS
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Available issues are temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": str(math.ceil(WORK_PAGE_TIMEOUT_SECONDS))},
|
|
)
|
|
result = _available_issue_page(items, page)
|
|
if stale:
|
|
result["stale"] = True
|
|
return JSONResponse(result)
|
|
|
|
|
|
async def _load_context_for_user(user_data: dict) -> dict:
|
|
repo_data, issues_data, prs_data = await asyncio.gather(
|
|
repos(), issues(), pull_requests()
|
|
)
|
|
if not all(field in user_data for field in ("id", "login")):
|
|
raise ContextPayloadError("Gitea current-user response did not include id and login")
|
|
if any(
|
|
data is not None and not isinstance(data, list)
|
|
for data in (repo_data, issues_data, prs_data)
|
|
):
|
|
raise ContextPayloadError("Gitea collection response was not a list")
|
|
return _context_payload(user_data, repo_data, issues_data, prs_data)
|
|
|
|
|
|
async def _build_live_snapshot(sections: set[str] | None = None) -> dict:
|
|
requested = set(sections or LIVE_SNAPSHOT_SECTIONS)
|
|
results: dict[str, object] = {}
|
|
user_data: dict | None = None
|
|
if requested & {"context", "events"}:
|
|
try:
|
|
user_data = await current_user()
|
|
if not isinstance(user_data, dict) or not user_data.get("login"):
|
|
raise ContextPayloadError("Gitea current-user response was invalid")
|
|
except Exception as exc:
|
|
for section in requested & {"context", "events"}:
|
|
results[section] = exc
|
|
|
|
loads: dict[str, Awaitable[Any]] = {}
|
|
if "context" in requested and "context" not in results:
|
|
assert user_data is not None
|
|
loads["context"] = _load_context_for_user(user_data)
|
|
if "events" in requested and "events" not in results:
|
|
assert user_data is not None
|
|
loads["events"] = activity_events(user_data)
|
|
if "notifications" in requested:
|
|
loads["notifications"] = notifications()
|
|
if loads:
|
|
loaded = await asyncio.gather(*loads.values(), return_exceptions=True)
|
|
results.update(zip(loads, loaded))
|
|
|
|
context_result = results.get("context")
|
|
events_result = results.get("events")
|
|
notifications_result = results.get("notifications")
|
|
context_ok = "context" in requested and not isinstance(context_result, BaseException)
|
|
events_ok = "events" in requested and not isinstance(events_result, BaseException)
|
|
notifications_ok = (
|
|
"notifications" in requested
|
|
and not isinstance(notifications_result, BaseException)
|
|
)
|
|
notification_items = notifications_result
|
|
notification_pagination = None
|
|
if notifications_ok and isinstance(notifications_result, dict):
|
|
notification_items = notifications_result.get("items")
|
|
notification_pagination = {
|
|
"page": notifications_result.get("page", 1),
|
|
"total": notifications_result.get("total", 0),
|
|
"has_more": notifications_result.get("has_more") is True,
|
|
}
|
|
notifications_ok = isinstance(notification_items, list)
|
|
return {
|
|
"context": context_result if context_ok else None,
|
|
"events": events_result if events_ok else None,
|
|
"notifications": notification_items if notifications_ok else None,
|
|
"notification_pagination": notification_pagination if notifications_ok else None,
|
|
"sections": {
|
|
section: "fresh" if ok else "temporarily unavailable"
|
|
for section, ok in (
|
|
("context", context_ok),
|
|
("events", events_ok),
|
|
("notifications", notifications_ok),
|
|
)
|
|
if section in requested
|
|
},
|
|
}
|
|
|
|
|
|
async def _build_live_snapshot_before_deadline(sections: set[str]) -> dict:
|
|
async with asyncio.timeout(CONTEXT_TIMEOUT_SECONDS):
|
|
return await _build_live_snapshot(sections)
|
|
|
|
|
|
def _record_live_section_failure(section: str) -> None:
|
|
_live_section_failure_count[section] += 1
|
|
delay = min(
|
|
LIVE_SNAPSHOT_RETRY_MAX_SECONDS,
|
|
LIVE_SNAPSHOT_RETRY_BASE_SECONDS
|
|
* (2 ** (_live_section_failure_count[section] - 1)),
|
|
)
|
|
_live_section_retry_at[section] = time.monotonic() + delay
|
|
|
|
|
|
def _due_live_sections(now: float) -> set[str]:
|
|
due = set()
|
|
for section in LIVE_SNAPSHOT_SECTIONS:
|
|
retry_at = _live_section_retry_at[section]
|
|
if retry_at is not None and now < retry_at:
|
|
continue
|
|
created_at = _live_section_created_at[section]
|
|
if created_at is None or now - created_at >= LIVE_SNAPSHOT_FRESHNESS_SECONDS:
|
|
due.add(section)
|
|
return due
|
|
|
|
|
|
def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
|
|
if previous is None:
|
|
return refreshed
|
|
merged = dict(previous)
|
|
sections = dict(previous.get("sections") or {})
|
|
for section, state in (refreshed.get("sections") or {}).items():
|
|
if state == "fresh":
|
|
merged[section] = refreshed.get(section)
|
|
sections[section] = "fresh"
|
|
if section == "notifications":
|
|
merged["notification_pagination"] = refreshed.get(
|
|
"notification_pagination"
|
|
)
|
|
elif previous.get(section) is not None:
|
|
sections[section] = "stale"
|
|
else:
|
|
merged[section] = None
|
|
sections[section] = "temporarily unavailable"
|
|
merged["sections"] = sections
|
|
return merged
|
|
|
|
|
|
async def _refresh_live_snapshot(sections: set[str]) -> dict:
|
|
global _live_snapshot_value, _live_snapshot_created_at
|
|
try:
|
|
refreshed = await _build_live_snapshot_before_deadline(sections)
|
|
except Exception:
|
|
for section in sections:
|
|
_record_live_section_failure(section)
|
|
raise
|
|
now = time.monotonic()
|
|
refreshed_states = refreshed.get("sections") or {}
|
|
for section in sections:
|
|
if refreshed_states.get(section) == "fresh":
|
|
_live_section_created_at[section] = now
|
|
_live_section_failure_count[section] = 0
|
|
_live_section_retry_at[section] = None
|
|
else:
|
|
_record_live_section_failure(section)
|
|
result = _merge_live_snapshot(_live_snapshot_value, refreshed)
|
|
result = _without_read_notifications(result)
|
|
_live_snapshot_value = result
|
|
successful_times = [value for value in _live_section_created_at.values() if value is not None]
|
|
_live_snapshot_created_at = max(successful_times) if successful_times else None
|
|
return result
|
|
|
|
|
|
def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
|
|
if task.cancelled():
|
|
return
|
|
task.exception()
|
|
|
|
|
|
def _start_live_snapshot_refresh(sections: set[str]) -> asyncio.Task:
|
|
global _live_snapshot_task
|
|
global _live_snapshot_refreshing_sections
|
|
if _live_snapshot_task is None or _live_snapshot_task.done():
|
|
_live_snapshot_refreshing_sections = set(sections)
|
|
_live_snapshot_task = asyncio.create_task(_refresh_live_snapshot(sections))
|
|
_live_snapshot_task.add_done_callback(_consume_live_snapshot_failure)
|
|
return _live_snapshot_task
|
|
|
|
|
|
def _live_snapshot_payload(value: dict, *, stale: bool, revalidating: bool) -> dict:
|
|
payload = dict(value)
|
|
now = time.monotonic()
|
|
section_freshness = {}
|
|
for section in LIVE_SNAPSHOT_SECTIONS:
|
|
created_at = _live_section_created_at[section]
|
|
retry_at = _live_section_retry_at[section]
|
|
age = max(0.0, now - created_at) if created_at is not None else 0.0
|
|
retry_in = max(0.0, retry_at - now) if retry_at is not None else 0.0
|
|
section_freshness[section] = {
|
|
"age_seconds": round(age, 3),
|
|
"stale": (value.get("sections") or {}).get(section) != "fresh"
|
|
or created_at is None
|
|
or age >= LIVE_SNAPSHOT_FRESHNESS_SECONDS,
|
|
"revalidating": revalidating
|
|
and section in _live_snapshot_refreshing_sections,
|
|
"degraded": retry_at is not None,
|
|
"retry_in_seconds": math.ceil(retry_in),
|
|
}
|
|
ages = [item["age_seconds"] for item in section_freshness.values()]
|
|
retries = [item["retry_in_seconds"] for item in section_freshness.values() if item["retry_in_seconds"]]
|
|
degraded = any(item["degraded"] for item in section_freshness.values())
|
|
payload["freshness"] = {
|
|
"age_seconds": max(ages, default=0.0),
|
|
"fresh_for_seconds": LIVE_SNAPSHOT_FRESHNESS_SECONDS,
|
|
"stale": stale or any(item["stale"] for item in section_freshness.values()),
|
|
"revalidating": revalidating,
|
|
"degraded": degraded,
|
|
"last_refresh_failed": degraded,
|
|
"retry_in_seconds": min(retries, default=0),
|
|
"sections": section_freshness,
|
|
}
|
|
return payload
|
|
|
|
|
|
def _remove_notification_from_live_snapshot(thread_id: int) -> None:
|
|
global _live_snapshot_value, _read_notification_ids
|
|
_read_notification_ids = _read_notification_ids | {thread_id}
|
|
if _live_snapshot_value is None:
|
|
return
|
|
retained_notifications = _live_snapshot_value.get("notifications")
|
|
if not isinstance(retained_notifications, list):
|
|
return
|
|
updated = dict(_live_snapshot_value)
|
|
updated["notifications"] = [
|
|
notification
|
|
for notification in retained_notifications
|
|
if not isinstance(notification, dict) or notification.get("id") != thread_id
|
|
]
|
|
_live_snapshot_value = updated
|
|
|
|
|
|
def _without_read_notifications(snapshot: dict) -> dict:
|
|
global _read_notification_ids
|
|
snapshot_notifications = snapshot.get("notifications")
|
|
if not isinstance(snapshot_notifications, list):
|
|
return snapshot
|
|
returned_ids = {
|
|
notification.get("id")
|
|
for notification in snapshot_notifications
|
|
if isinstance(notification, dict)
|
|
}
|
|
updated = dict(snapshot)
|
|
updated["notifications"] = [
|
|
notification
|
|
for notification in snapshot_notifications
|
|
if not isinstance(notification, dict)
|
|
or notification.get("id") not in _read_notification_ids
|
|
]
|
|
_read_notification_ids = _read_notification_ids.intersection(returned_ids)
|
|
return updated
|
|
|
|
|
|
@app.get("/api/v1/live")
|
|
async def live_snapshot() -> JSONResponse:
|
|
"""Return a freshness-bounded snapshot and share identical upstream loads."""
|
|
global _live_snapshot_task, _live_snapshot_value, _live_snapshot_created_at
|
|
now = time.monotonic()
|
|
due_sections = _due_live_sections(now)
|
|
if (
|
|
_live_snapshot_value is not None
|
|
and not due_sections
|
|
):
|
|
return JSONResponse(
|
|
_live_snapshot_payload(
|
|
_live_snapshot_value,
|
|
stale=any(
|
|
state != "fresh"
|
|
for state in _live_snapshot_value.get("sections", {}).values()
|
|
),
|
|
revalidating=False,
|
|
)
|
|
)
|
|
task = _start_live_snapshot_refresh(due_sections)
|
|
if _live_snapshot_value is not None:
|
|
return JSONResponse(
|
|
_live_snapshot_payload(
|
|
_live_snapshot_value, stale=True, revalidating=True
|
|
)
|
|
)
|
|
try:
|
|
result = await asyncio.shield(task)
|
|
return JSONResponse(
|
|
_live_snapshot_payload(result, stale=False, revalidating=False)
|
|
)
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
|
|
status_code=503,
|
|
headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Gitea live snapshot is temporarily unavailable"},
|
|
status_code=503,
|
|
)
|
|
|
|
|
|
|
|
@app.get("/api/v1/events")
|
|
async def event_stream():
|
|
try:
|
|
return await asyncio.wait_for(
|
|
activity_events(), timeout=EVENT_STREAM_TIMEOUT_SECONDS
|
|
)
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{
|
|
"error": (
|
|
"Gitea event stream request timed out after "
|
|
f"{EVENT_STREAM_TIMEOUT_SECONDS:g}s"
|
|
)
|
|
},
|
|
status_code=503,
|
|
headers={
|
|
"Retry-After": str(
|
|
max(1, math.ceil(EVENT_STREAM_TIMEOUT_SECONDS))
|
|
)
|
|
},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Gitea event stream is temporarily unavailable"},
|
|
status_code=503,
|
|
headers={
|
|
"Retry-After": str(
|
|
max(1, math.ceil(EVENT_STREAM_TIMEOUT_SECONDS))
|
|
)
|
|
},
|
|
)
|
|
|
|
|
|
async def _mark_notification_read_result(thread_id: int) -> tuple[int, bool]:
|
|
try:
|
|
await asyncio.wait_for(
|
|
mark_notification_read(thread_id),
|
|
timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
|
|
)
|
|
except Exception:
|
|
return thread_id, False
|
|
_remove_notification_from_live_snapshot(thread_id)
|
|
return thread_id, True
|
|
|
|
|
|
@app.patch("/api/v1/notifications/read")
|
|
async def read_notifications(batch: NotificationReadBatch) -> JSONResponse:
|
|
thread_ids = list(dict.fromkeys(batch.ids))
|
|
|
|
semaphore = asyncio.Semaphore(BULK_NOTIFICATION_CONCURRENCY)
|
|
|
|
async def mark_within_limit(thread_id: int) -> tuple[int, bool]:
|
|
async with semaphore:
|
|
return await _mark_notification_read_result(thread_id)
|
|
|
|
tasks = [asyncio.create_task(mark_within_limit(thread_id)) for thread_id in thread_ids]
|
|
done, pending = await asyncio.wait(
|
|
tasks, timeout=BULK_NOTIFICATION_DEADLINE_SECONDS
|
|
)
|
|
for task in pending:
|
|
task.cancel()
|
|
if pending:
|
|
await asyncio.gather(*pending, return_exceptions=True)
|
|
|
|
succeeded_ids = {
|
|
thread_id
|
|
for task in done
|
|
if not task.cancelled() and task.exception() is None
|
|
for thread_id, succeeded in [task.result()]
|
|
if succeeded
|
|
}
|
|
failed = [thread_id for thread_id in thread_ids if thread_id not in succeeded_ids]
|
|
return JSONResponse(
|
|
{
|
|
"marked": [thread_id for thread_id in thread_ids if thread_id in succeeded_ids],
|
|
"failed": failed,
|
|
},
|
|
headers={"Retry-After": "1"} if failed else None,
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/notifications")
|
|
async def notification_page(page: int = Query(default=1, ge=1)) -> JSONResponse:
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.notification_page(page),
|
|
timeout=NOTIFICATION_PAGE_TIMEOUT_SECONDS,
|
|
)
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": "Unread updates timed out. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Unread updates are temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.get("/api/v1/notifications/{thread_id}")
|
|
async def notification_thread_detail(
|
|
thread_id: int = PathParam(gt=0),
|
|
) -> JSONResponse:
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
notification_detail(thread_id),
|
|
timeout=NOTIFICATION_DETAIL_TIMEOUT_SECONDS,
|
|
)
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": "Loading the update timed out. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The update is temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.patch("/api/v1/notifications/{thread_id}/read")
|
|
async def read_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
|
|
try:
|
|
await asyncio.wait_for(
|
|
mark_notification_read(thread_id),
|
|
timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
|
|
)
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": "Marking the update read timed out. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The update could not be marked read. Please retry."},
|
|
status_code=503,
|
|
)
|
|
_remove_notification_from_live_snapshot(thread_id)
|
|
return JSONResponse({"id": thread_id, "status": "read"})
|
|
|
|
|
|
@app.post("/api/v1/notifications/{thread_id}/reply", status_code=201)
|
|
async def reply_to_notification(
|
|
reply: NotificationReply,
|
|
thread_id: int = PathParam(gt=0),
|
|
idempotency_key: str | None = Header(default=None, max_length=128),
|
|
) -> JSONResponse:
|
|
try:
|
|
result = await _run_idempotent_authored_action(
|
|
gitea_proxy.reply_to_notification(thread_id, reply.body),
|
|
idempotency_key=idempotency_key,
|
|
fingerprint=("notification-reply", thread_id, reply.body),
|
|
timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{
|
|
"error": (
|
|
"The reply could not be posted. Your draft is safe; please retry."
|
|
)
|
|
},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result, status_code=201)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/claim")
|
|
async def claim_available_issue(
|
|
owner: str, repo: str, number: int = PathParam(gt=0)
|
|
) -> JSONResponse:
|
|
repository = f"{owner}/{repo}"
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.claim_available_issue(repository, number),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except gitea_proxy.IssueNotAvailableError:
|
|
return JSONResponse(
|
|
{"error": "This issue was already claimed or is no longer open. Refresh Find Work."},
|
|
status_code=409,
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The issue could not be assigned. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
if _available_issue_snapshot_value is not None:
|
|
_available_issue_snapshot_value[:] = [
|
|
item for item in _available_issue_snapshot_value
|
|
if not (
|
|
item.get("repository") == repository and item.get("number") == number
|
|
)
|
|
]
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/release")
|
|
async def release_assigned_issue(
|
|
owner: str, repo: str, number: int = PathParam(gt=0)
|
|
) -> JSONResponse:
|
|
global _available_issue_snapshot_value, _available_issue_snapshot_created_at
|
|
repository = f"{owner}/{repo}"
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.release_assigned_issue(repository, number),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except gitea_proxy.IssueNotAvailableError:
|
|
raise HTTPException(status_code=404, detail="Assigned issue 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"},
|
|
)
|
|
if result.get("available"):
|
|
_available_issue_snapshot_value = None
|
|
_available_issue_snapshot_created_at = None
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/detail")
|
|
async def assigned_issue_detail(owner: str, repo: str, number: int = PathParam(gt=0)):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_assigned_issue():
|
|
if not await gitea_proxy.is_assigned_issue(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
return await gitea_proxy.issue_detail(repository, number)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
load_assigned_issue(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": "Loading the issue timed out. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The issue is temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/content")
|
|
async def update_assigned_issue_content(
|
|
update: IssueContentUpdate,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.update_assigned_issue(
|
|
repository,
|
|
number,
|
|
update.title,
|
|
update.body,
|
|
update.expected_updated_at,
|
|
),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except gitea_proxy.IssueEditConflictError:
|
|
return JSONResponse(
|
|
{"error": "This issue changed in Gitea. Your draft is safe; reload the latest issue before saving."},
|
|
status_code=409,
|
|
)
|
|
except gitea_proxy.IssueNotAvailableError:
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The issue could not be updated. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/due-date")
|
|
async def update_assigned_issue_due_date(
|
|
update: IssueDueDateUpdate,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.update_assigned_issue_due_date(
|
|
repository, number, update.due_date
|
|
),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except gitea_proxy.IssueNotAvailableError:
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The due date could not be updated. Your selection is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/milestone")
|
|
async def update_assigned_issue_milestone(
|
|
update: IssueMilestoneUpdate,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
gitea_proxy.update_assigned_issue_milestone(
|
|
repository, number, update.milestone_id
|
|
),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except gitea_proxy.IssueNotAvailableError:
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
except ValueError as exc:
|
|
if str(exc) == "Unknown open repository milestone":
|
|
raise HTTPException(status_code=422, detail=str(exc))
|
|
return JSONResponse(
|
|
{"error": "The milestone could not be updated. Your selection is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The milestone could not be updated. Your selection is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/milestones")
|
|
async def repository_milestones(owner: str, repo: str):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_milestones():
|
|
available = await gitea_proxy.repos()
|
|
accessible = {
|
|
item.get("full_name")
|
|
for item in (available if isinstance(available, list) else [])
|
|
if isinstance(item, dict)
|
|
}
|
|
if repository not in accessible:
|
|
raise HTTPException(status_code=404, detail="Repository not found")
|
|
return await gitea_proxy.repo_milestones(repository)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
load_milestones(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Milestones could not be loaded. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/labels")
|
|
async def repository_labels(owner: str, repo: str):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_labels():
|
|
available = await gitea_proxy.repos()
|
|
accessible = {
|
|
item.get("full_name")
|
|
for item in (available if isinstance(available, list) else [])
|
|
if isinstance(item, dict)
|
|
}
|
|
if repository not in accessible:
|
|
raise HTTPException(status_code=404, detail="Repository not found")
|
|
return await gitea_proxy.repo_labels(repository)
|
|
|
|
try:
|
|
return await asyncio.wait_for(load_labels(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Labels could not be loaded. Issue creation is still available."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.post("/api/v1/repos/{owner}/{repo}/issues", status_code=201)
|
|
async def create_assigned_issue(
|
|
creation: IssueCreation,
|
|
owner: str,
|
|
repo: str,
|
|
idempotency_key: str | None = Header(default=None, max_length=128),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def create_issue():
|
|
user, available = await asyncio.gather(
|
|
gitea_proxy.current_user(), gitea_proxy.repos()
|
|
)
|
|
login = user.get("login") if isinstance(user, dict) else None
|
|
accessible = {
|
|
item.get("full_name")
|
|
for item in (available if isinstance(available, list) else [])
|
|
if isinstance(item, dict)
|
|
}
|
|
if not login or repository not in accessible:
|
|
raise HTTPException(status_code=404, detail="Repository not found")
|
|
if creation.label_ids:
|
|
available_labels = await gitea_proxy.repo_labels(repository)
|
|
valid_label_ids = {
|
|
item.get("id")
|
|
for item in available_labels
|
|
if isinstance(item, dict) and isinstance(item.get("id"), int)
|
|
}
|
|
if any(label_id not in valid_label_ids for label_id in creation.label_ids):
|
|
raise HTTPException(status_code=422, detail="Unknown repository label")
|
|
return await gitea_proxy.create_issue(
|
|
repository, creation.title, creation.body, login, creation.label_ids
|
|
)
|
|
|
|
try:
|
|
result = await _run_idempotent_authored_action(
|
|
create_issue(),
|
|
idempotency_key=idempotency_key,
|
|
fingerprint=(
|
|
"issue-create",
|
|
repository,
|
|
creation.title,
|
|
creation.body,
|
|
tuple(creation.label_ids),
|
|
),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The issue could not be created. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result, status_code=201)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/comments")
|
|
async def assigned_issue_conversation(
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
page: int | None = Query(default=None, ge=1, le=100),
|
|
limit: int = Query(default=20, ge=1, le=50),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_conversation():
|
|
if not await gitea_proxy.is_assigned_issue(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
return await gitea_proxy.issue_conversation_page(repository, number, page, limit)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
load_conversation(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The conversation could not be loaded. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/comments", status_code=201)
|
|
async def comment_on_assigned_issue(
|
|
comment: IssueComment,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
idempotency_key: str | None = Header(default=None, max_length=128),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def post_comment():
|
|
if not await gitea_proxy.is_assigned_issue(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
return await gitea_proxy.comment_on_issue(repository, number, comment.body)
|
|
|
|
try:
|
|
result = await _run_idempotent_authored_action(
|
|
post_comment(),
|
|
idempotency_key=idempotency_key,
|
|
fingerprint=("issue-comment", repository, number, comment.body),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The comment could not be posted. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result, status_code=201)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/close")
|
|
async def close_assigned_issue(owner: str, repo: str, number: int = PathParam(gt=0)):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def close_issue():
|
|
if not await gitea_proxy.is_assigned_issue(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
return await gitea_proxy.close_issue(repository, number)
|
|
|
|
try:
|
|
return await asyncio.wait_for(close_issue(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The issue could not be closed. It remains in My Work; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/labels")
|
|
async def assigned_issue_label_options(
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_labels():
|
|
if not await gitea_proxy.is_assigned_issue(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
return await gitea_proxy.repo_labels(repository)
|
|
|
|
try:
|
|
return await asyncio.wait_for(load_labels(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Labels could not be loaded. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/labels")
|
|
async def update_assigned_issue_labels(
|
|
update: IssueLabelUpdate,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def update_labels():
|
|
if not await gitea_proxy.is_assigned_issue(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned issue not found")
|
|
available = await gitea_proxy.repo_labels(repository)
|
|
valid_ids = {
|
|
item.get("id") for item in available
|
|
if isinstance(item, dict) and isinstance(item.get("id"), int)
|
|
}
|
|
if any(label_id not in valid_ids for label_id in update.label_ids):
|
|
raise HTTPException(status_code=422, detail="Unknown repository label")
|
|
return await gitea_proxy.update_issue_labels(repository, number, update.label_ids)
|
|
|
|
try:
|
|
return await asyncio.wait_for(update_labels(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Labels could not be updated. Your selection is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/review")
|
|
async def review_detail(owner: str, repo: str, number: int):
|
|
async def load_requested_review():
|
|
repository = f"{owner}/{repo}"
|
|
if not await is_requested_review(repository, number):
|
|
raise HTTPException(status_code=404, detail="Review request not found")
|
|
return await pull_review_detail(repository, number)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
load_requested_review(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": "Pull request review details timed out. Please retry."},
|
|
status_code=503,
|
|
headers={
|
|
"Retry-After": str(
|
|
max(1, math.ceil(REVIEW_DETAIL_TIMEOUT_SECONDS))
|
|
)
|
|
},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "Pull request review details are temporarily unavailable"},
|
|
status_code=503,
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/detail")
|
|
async def assigned_pull_detail(owner: str, repo: str, number: int = PathParam(gt=0)):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_assigned_pull():
|
|
if not await gitea_proxy.is_assigned_pull(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned pull request not found")
|
|
return await gitea_proxy.pull_completion_detail(repository, number)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
load_assigned_pull(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except TimeoutError:
|
|
return JSONResponse(
|
|
{"error": "Loading the pull request timed out. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The pull request is temporarily unavailable. Please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments")
|
|
async def assigned_pull_conversation(
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
page: int | None = Query(default=None, ge=1, le=100),
|
|
limit: int = Query(default=20, ge=1, le=50),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def load_conversation():
|
|
if not await gitea_proxy.is_assigned_pull(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned pull request not found")
|
|
return await gitea_proxy.issue_conversation_page(repository, number, page, limit)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
load_conversation(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The conversation could not be loaded. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments", status_code=201)
|
|
async def comment_on_assigned_pull(
|
|
comment: IssueComment,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
idempotency_key: str | None = Header(default=None, max_length=128),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def post_comment():
|
|
if not await gitea_proxy.is_assigned_pull(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned pull request not found")
|
|
return await gitea_proxy.comment_on_issue(repository, number, comment.body)
|
|
|
|
try:
|
|
result = await _run_idempotent_authored_action(
|
|
post_comment(),
|
|
idempotency_key=idempotency_key,
|
|
fingerprint=("pull-comment", repository, number, comment.body),
|
|
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The comment could not be posted. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result, status_code=201)
|
|
|
|
|
|
@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/merge")
|
|
async def merge_assigned_pull(
|
|
submission: PullMergeSubmission,
|
|
owner: str,
|
|
repo: str,
|
|
number: int = PathParam(gt=0),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def merge_pull():
|
|
if not await gitea_proxy.is_assigned_pull(repository, number):
|
|
raise HTTPException(status_code=404, detail="Assigned pull request not found")
|
|
return await gitea_proxy.merge_assigned_pull(
|
|
repository, number, submission.expected_head_sha
|
|
)
|
|
|
|
try:
|
|
return await asyncio.wait_for(
|
|
merge_pull(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except gitea_proxy.StalePullError:
|
|
return JSONResponse(
|
|
{"error": "New commits were pushed. Refresh before merging."},
|
|
status_code=409,
|
|
)
|
|
except gitea_proxy.PullNotMergeableError:
|
|
return JSONResponse(
|
|
{"error": "This pull request is not currently safe to merge. Refresh its status."},
|
|
status_code=409,
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The pull request could not be merged. It remains in My Work; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
|
|
|
|
@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201)
|
|
async def submit_review(
|
|
submission: PullReviewSubmission,
|
|
owner: str,
|
|
repo: str,
|
|
number: int,
|
|
idempotency_key: str | None = Header(default=None, max_length=128),
|
|
):
|
|
repository = f"{owner}/{repo}"
|
|
|
|
async def submit_requested_review():
|
|
if not await is_requested_review(repository, number):
|
|
raise HTTPException(status_code=404, detail="Review request not found")
|
|
args = (
|
|
repository,
|
|
number,
|
|
submission.expected_head_sha,
|
|
submission.decision,
|
|
submission.body,
|
|
)
|
|
if submission.comments:
|
|
return await gitea_proxy.submit_pull_review(
|
|
*args,
|
|
[comment.model_dump() for comment in submission.comments],
|
|
)
|
|
return await gitea_proxy.submit_pull_review(*args)
|
|
|
|
try:
|
|
comment_fingerprint = tuple(
|
|
(comment.path, comment.body, comment.new_position, comment.old_position)
|
|
for comment in submission.comments
|
|
)
|
|
result = await _run_idempotent_authored_action(
|
|
submit_requested_review(),
|
|
idempotency_key=idempotency_key,
|
|
fingerprint=(
|
|
"pull-review", repository, number, submission.expected_head_sha,
|
|
submission.decision, submission.body, comment_fingerprint,
|
|
),
|
|
timeout=REVIEW_DETAIL_TIMEOUT_SECONDS,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except gitea_proxy.StaleReviewError:
|
|
return JSONResponse(
|
|
{"error": "New commits were pushed. Refresh the review before submitting."},
|
|
status_code=409,
|
|
)
|
|
except gitea_proxy.InvalidReviewCommentError:
|
|
return JSONResponse(
|
|
{"error": "An inline comment no longer matches this pull request. Refresh the review."},
|
|
status_code=422,
|
|
)
|
|
except Exception:
|
|
return JSONResponse(
|
|
{"error": "The review could not be submitted. Your draft is safe; please retry."},
|
|
status_code=503,
|
|
headers={"Retry-After": "1"},
|
|
)
|
|
return JSONResponse(result, status_code=201)
|