mirror of
https://github.com/OpenHands/OpenHands.git
synced 2026-03-22 13:47:19 +08:00
145 lines
4.4 KiB
Python
145 lines
4.4 KiB
Python
import httpx
|
|
import requests
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from openhands.server.auth import get_github_token
|
|
from openhands.utils.async_utils import call_sync_from_async
|
|
|
|
app = APIRouter(prefix='/api/github')
|
|
|
|
|
|
def require_github_token(request: Request):
|
|
github_token = get_github_token(request)
|
|
if not github_token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail='Missing GitHub token',
|
|
)
|
|
|
|
return github_token
|
|
|
|
|
|
@app.get('/repositories')
|
|
async def get_github_repositories(
|
|
page: int = 1,
|
|
per_page: int = 10,
|
|
sort: str = 'pushed',
|
|
installation_id: int | None = None,
|
|
github_token: str = Depends(require_github_token),
|
|
):
|
|
params: dict[str, str] = {
|
|
'page': str(page),
|
|
'per_page': str(per_page),
|
|
}
|
|
if installation_id:
|
|
github_api_url = (
|
|
f'https://api.github.com/user/installations/{installation_id}/repositories'
|
|
)
|
|
else:
|
|
github_api_url = 'https://api.github.com/user/repos'
|
|
params['sort'] = sort
|
|
|
|
headers = generate_github_headers(github_token)
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(github_api_url, headers=headers, params=params)
|
|
response.raise_for_status() # Raise an error for HTTP codes >= 400
|
|
json_response = JSONResponse(content=response.json())
|
|
|
|
# Forward the Link header if it exists
|
|
if 'Link' in response.headers:
|
|
json_response.headers['Link'] = response.headers['Link']
|
|
|
|
return json_response
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
raise HTTPException(
|
|
status_code=response.status_code if response else 500,
|
|
detail=f'Error fetching repositories: {str(e)}',
|
|
)
|
|
|
|
|
|
@app.get('/user')
|
|
async def get_github_user(github_token: str = Depends(require_github_token)):
|
|
headers = generate_github_headers(github_token)
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get('https://api.github.com/user', headers=headers)
|
|
response.raise_for_status() # Raise an error for HTTP codes >= 400
|
|
json_response = JSONResponse(content=response.json())
|
|
|
|
return json_response
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
raise HTTPException(
|
|
status_code=response.status_code if response else 500,
|
|
detail=f'Error fetching user: {str(e)}',
|
|
)
|
|
|
|
|
|
@app.get('/installations')
|
|
async def get_github_installation_ids(
|
|
github_token: str = Depends(require_github_token),
|
|
):
|
|
headers = generate_github_headers(github_token)
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
'https://api.github.com/user/installations', headers=headers
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
ids = [installation['id'] for installation in data['installations']]
|
|
return JSONResponse(content=ids)
|
|
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(
|
|
status_code=e.response.status_code if hasattr(e, 'response') else 500,
|
|
detail=f'Error fetching installations: {str(e)}',
|
|
)
|
|
|
|
|
|
@app.get('/search/repositories')
|
|
async def search_github_repositories(
|
|
query: str,
|
|
per_page: int = 5,
|
|
sort: str = 'stars',
|
|
order: str = 'desc',
|
|
github_token: str = Depends(require_github_token),
|
|
):
|
|
headers = generate_github_headers(github_token)
|
|
params = {
|
|
'q': query,
|
|
'per_page': per_page,
|
|
'sort': sort,
|
|
'order': order,
|
|
}
|
|
|
|
try:
|
|
response = await call_sync_from_async(
|
|
requests.get,
|
|
'https://api.github.com/search/repositories',
|
|
headers=headers,
|
|
params=params,
|
|
)
|
|
response.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
raise HTTPException(
|
|
status_code=response.status_code if response else 500,
|
|
detail=f'Error searching repositories: {str(e)}',
|
|
)
|
|
|
|
json_response = JSONResponse(content=response.json())
|
|
response.close()
|
|
|
|
return json_response
|
|
|
|
|
|
def generate_github_headers(token: str) -> dict[str, str]:
|
|
return {
|
|
'Authorization': f'Bearer {token}',
|
|
'Accept': 'application/vnd.github.v3+json',
|
|
}
|