feat: add explicit searchable repository picker (Closes #487)
CI / lint (pull_request) Successful in 1m2s
CI / build-release (pull_request) Successful in 6s
CI / release-candidate (pull_request) Has been skipped

This commit is contained in:
timmy committed 2026-08-10 14:01:49 +00:00
1 parent b1b3f758e4
commit 81e1af7f16
9 files changed
+277 -9

No files matched your search

+15
View File
@@ -301,6 +301,21 @@ async def repos() -> RepositoryItems:
)
async def search_repositories(query: str, limit: int = 20) -> list[dict]:
"""Search repositories visible to the authenticated Gitea user."""
response = await _get_client().get(
"/api/v1/repos/search",
headers=_auth(),
params={"q": query.strip(), "limit": limit, "page": 1},
)
response.raise_for_status()
payload = response.json()
items = payload.get("data") if isinstance(payload, dict) else None
if not isinstance(items, list):
raise ValueError("Gitea repository search response was not a list")
return [item for item in items if isinstance(item, dict)]
async def repository_access(repository: str) -> dict | None:
"""Return a repository only when the authenticated user can access it."""
response = await _get_client().get(
+29
View File
@@ -1644,6 +1644,35 @@ async def context() -> JSONResponse:
return JSONResponse(_context_payload(user_data, repo_data, issues_data, prs_data))
@app.get("/api/v1/repositories/search")
async def repository_search(
q: str = Query(min_length=2, max_length=80),
limit: int = Query(default=20, ge=1, le=20),
) -> JSONResponse:
"""Return minimal repository identities matching a bounded visible-repo search."""
query = q.strip()
if len(query) < 2:
return JSONResponse({"error": "Enter at least 2 search characters."}, status_code=400)
try:
matches = await asyncio.wait_for(
gitea_proxy.search_repositories(query, limit),
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
)
except Exception:
return JSONResponse(
{"error": "Repository search is temporarily unavailable. Please retry."},
status_code=503,
headers={"Retry-After": "1"},
)
items = [
{"id": item["id"], "name": item["name"], "full_name": item["full_name"]}
for item in matches
if isinstance(item, dict)
and all(field in item for field in ("id", "name", "full_name"))
]
return JSONResponse({"items": items}, headers={"Cache-Control": "no-store"})
@app.get("/api/v1/repositories")
async def repository_page(
page: int = Query(default=1, ge=1, le=1000),