0ce879aa44
- Implemented HTML pages for datasets, models, training, testing, and results. - Created API endpoints for managing repositories, results, tests, and training sessions. - Added functionality for streaming training progress via Server-Sent Events (SSE). - Introduced a Dockerfile for the ML runner with necessary dependencies. - Developed an SDK for user code execution within the runner container. - Enhanced CSS styles for improved UI layout and navigation. - Established a layout template for consistent HTML structure across pages. - Added JavaScript for dynamic interactions on the models page. - Implemented WebSocket handling for real-time communication with kiosk devices and controllers. - Implemented model registration and management API at /api/models - Added Gitea proxy API for repository interactions at /api/repos - Created results API for listing and comparing training results at /api/results - Developed training management API for enqueueing and retrieving training jobs at /api/trainings - Introduced SSE endpoint for live training progress updates - Added HTML pages for models, datasets, and training management - Created a Dockerfile for the ML runner with necessary dependencies - Developed SDK for user code execution within the runner container - Enhanced CSS styles for improved UI/UX - Implemented WebSocket communication for real-time device and controller interactions in the kiosk system
58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
"""Client Gitea: browse repo, branches, commits, file raw, clone URL autenticato."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from core.config import settings
|
|
|
|
|
|
def _headers() -> dict:
|
|
h = {"Accept": "application/json"}
|
|
if settings.gitea_token:
|
|
h["Authorization"] = f"token {settings.gitea_token}"
|
|
return h
|
|
|
|
|
|
def clone_url(owner_repo: str) -> str:
|
|
"""URL https://oauth2:TOKEN@<host>/owner/repo.git — usato SOLO lato server."""
|
|
if not settings.gitea_url:
|
|
raise RuntimeError("GITEA_URL not configured")
|
|
base = settings.gitea_url.rstrip("/")
|
|
if settings.gitea_token:
|
|
base = base.replace("https://", f"https://oauth2:{settings.gitea_token}@").replace(
|
|
"http://", f"http://oauth2:{settings.gitea_token}@"
|
|
)
|
|
return f"{base}/{owner_repo}.git"
|
|
|
|
|
|
async def _get(path: str, params: Optional[dict] = None) -> list | dict:
|
|
url = f"{settings.gitea_url.rstrip('/')}/api/v1{path}"
|
|
async with httpx.AsyncClient(timeout=15.0) as c:
|
|
r = await c.get(url, params=params, headers=_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def list_repos(limit: int = 50) -> list[dict]:
|
|
data = await _get("/repos/search", params={"limit": str(limit)})
|
|
return data.get("data", []) if isinstance(data, dict) else []
|
|
|
|
|
|
async def list_branches(owner_repo: str) -> list[dict]:
|
|
return await _get(f"/repos/{owner_repo}/branches")
|
|
|
|
|
|
async def list_commits(owner_repo: str, branch: str = "main", limit: int = 50) -> list[dict]:
|
|
return await _get(f"/repos/{owner_repo}/commits", params={"sha": branch, "limit": str(limit)})
|
|
|
|
|
|
async def get_file_raw(owner_repo: str, ref: str, path: str) -> bytes:
|
|
"""Scarica il file raw alla revisione indicata."""
|
|
url = f"{settings.gitea_url.rstrip('/')}/api/v1/repos/{owner_repo}/raw/{path}"
|
|
async with httpx.AsyncClient(timeout=15.0) as c:
|
|
r = await c.get(url, params={"ref": ref}, headers=_headers())
|
|
r.raise_for_status()
|
|
return r.content
|