devplacepy/index.py at 6f340a6818ebe1e5334a055c431f3ec9b43d11af

 # retoor <retoor@molodetz.nl>
 import logging
 from fastapi import APIRouter, Request
 from fastapi.responses import HTMLResponse
 from devplacepy.attachments import get_attachments, get_attachments_batch
 from devplacepy.database import (
     build_pagination,
     get_users_by_uids,
     get_blocked_uids,
     mark_notifications_read_by_target,
 )
 from devplacepy.responses import respond
 from devplacepy.schemas import IssueDetailOut, IssuesOut
 from devplacepy.services.gitea import runtime, store
 from devplacepy.services.gitea.client import STATE_OPEN, GiteaError
 from devplacepy.services.gitea.config import gitea_config
 from devplacepy.utils import get_current_user, is_admin, not_found
 from devplacepy.routers.issues._shared import (
     PER_PAGE,
     issue_seo,
     comment_item,
     issue_item,
     state_or_default,
     tracker_unavailable,
 )
 logger = logging.getLogger(__name__)
 router = APIRouter()
 @router.get("", response_class=HTMLResponse)
 async def issues_page(request: Request, state: str = STATE_OPEN, page: int = 1):
     user = get_current_user(request)
     state = state_or_default(state)
     page = max(1, page)
     seo_ctx = issue_seo(
         request,
         title="Issue Reports",
         description="Report issues and track their progress on DevPlace.",
     )
     base_ctx = {
         **seo_ctx,
         "request": request,
         "user": user,
         "state": state,
         "issues": [],
         "pagination": None,
         "configured": True,
         "error_message": None,
         "viewer_is_admin": is_admin(user),
     }
     config = gitea_config()
     if not config.is_configured:
         base_ctx["configured"] = False
         base_ctx["error_message"] = "The issue tracker is not configured yet."
         return respond(request, "issues.html", base_ctx, model=IssuesOut)
     try:
         issues, total = await runtime.get_client().list_issues(
             state=state, page=page, limit=PER_PAGE
         )
     except GiteaError as exc:
         logger.warning("Could not load issue list: %s", exc)
         base_ctx["error_message"] = "The issue tracker is temporarily unavailable."
         return respond(request, "issues.html", base_ctx, model=IssuesOut)
     numbers = [int(issue.get("number", 0)) for issue in issues]
     author_by_number = store.author_map(numbers)
     blocked = get_blocked_uids(user["uid"]) if user else frozenset()
     if blocked:
         issues = [
             issue
             for issue in issues
             if author_by_number.get(int(issue.get("number", 0))) not in blocked
         ]
     users_map = get_users_by_uids([uid for uid in author_by_number.values() if uid])
     base_ctx["issues"] = [
         issue_item(issue, author_by_number.get(int(issue.get("number", 0))), users_map)
         for issue in issues
     ]
     base_ctx["pagination"] = build_pagination(page, total, PER_PAGE)
     return respond(request, "issues.html", base_ctx, model=IssuesOut)
 @router.get("/{number:int}", response_class=HTMLResponse)
 async def issue_detail(request: Request, number: int):
     user = get_current_user(request)
     if not gitea_config().is_configured:
         raise not_found("Issue not found")
     client = runtime.get_client()
     try:
         issue = await client.get_issue(number)
     except GiteaError as exc:
         if exc.status == 404:
             raise not_found("Issue not found")
         logger.warning("Could not load issue #%s: %s", number, exc)
         return tracker_unavailable(request)
     try:
         comments = await client.list_comments(number)
     except GiteaError as exc:
         logger.warning("Could not load comments for issue #%s: %s", number, exc)
         comments = []
     if user:
         mark_notifications_read_by_target(user["uid"], f"/issues?highlight={number}")
     author_uid = store.author_uid_for_issue(number)
     comment_authors = store.comment_author_map(
         [int(comment.get("id", 0)) for comment in comments]
     )
     uids = [uid for uid in [author_uid, *comment_authors.values()] if uid]
     users_map = get_users_by_uids(uids)
     body = issue.get("body", "")
     issue = issue_item(issue, author_uid, users_map)
     is_open = issue.get("state") == STATE_OPEN
     def _decorate(rows):
         items = []
         for row in rows:
             att = dict(row)
             att["can_modify"] = bool(
                 user
                 and is_open
                 and (att.get("user_uid") == user["uid"] or is_admin(user))
             )
             items.append(att)
         return items
     issue_attachments = _decorate(get_attachments("issue", str(number)))
     comment_attachments = get_attachments_batch(
         "issue_comment", [str(int(comment.get("id", 0))) for comment in comments]
     )
     comment_items = []
     for comment in comments:
         item = comment_item(
             comment, comment_authors.get(int(comment.get("id", 0))), users_map
         )
         item["attachments"] = _decorate(
             comment_attachments.get(str(item["id"]), [])
         )
         comment_items.append(item)
     seo_ctx = issue_seo(
         request,
         title=f"Issue #{number}: {issue.get('title', '')}",
         description=body or issue.get("title", ""),
         seo_target=("issue", str(number)),
     )
     return respond(
         request,
         "issue_detail.html",
         {
             **seo_ctx,
             "request": request,
             "user": user,
             "issue": issue,
             "body": body,
             "comments": comment_items,
             "attachments": issue_attachments,
             "can_comment": user is not None,
             "can_attach": user is not None and is_open,
             "viewer_is_admin": is_admin(user),
         },
         model=IssueDetailOut,
     )