9901a61a0c
Some checks failed
Check Python Version Consistency / Check Python Version (push) Has been cancelled
51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
import os
|
|
import httpx
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _gitea_headers() -> dict:
|
|
"""Constructs headers for Gitea API requests."""
|
|
return {"Authorization": f"token {os.environ.get('GITEA_TOKEN')}", "Accept": "application/json"}
|
|
|
|
async def handle_get_file_content(p: dict, default_repo: str) -> dict:
|
|
"""Gets the raw content of a file from the Gitea repository."""
|
|
gitea_url = os.environ.get("GITEA_URL")
|
|
if not gitea_url:
|
|
raise ValueError("GITEA_URL environment variable is not set.")
|
|
|
|
path = p.get("path")
|
|
if not path:
|
|
raise ValueError("Missing required parameter: 'path' for get_file_content")
|
|
|
|
repo_id = p.get("repo", default_repo)
|
|
ref = p.get("ref", "main")
|
|
|
|
url = f"{gitea_url}/api/v1/repos/{repo_id}/raw/{path}?ref={ref}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(url, headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
return {"path": path, "content": r.text, "encoding": "text"}
|
|
|
|
async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
|
|
"""Lists files and directories in a given path in the Gitea repository."""
|
|
gitea_url = os.environ.get("GITEA_URL")
|
|
if not gitea_url:
|
|
raise ValueError("GITEA_URL environment variable is not set.")
|
|
|
|
path = p.get("path", "")
|
|
repo_id = p.get("repo", default_repo)
|
|
ref = p.get("ref", "main")
|
|
|
|
url = f"{gitea_url}/api/v1/repos/{repo_id}/contents/{path}?ref={ref}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(url, headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
files = r.json()
|
|
return {
|
|
"path": path,
|
|
"files": [{"name": f.get("name"), "type": f.get("type"), "path": f.get("path")} for f in files]
|
|
}
|