import html from urllib.parse import quote from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse from ..auth import require_user from ..config import ( DEFAULT_APPFACTORY_HOST, DEFAULT_GITEA_ORG, DELETE_APP_SCRIPT, DEPLOY_SCRIPT, GENERATE_COMPOSE_SCRIPT, NEW_APP_SCRIPT, read_env_value, ) from ..db.apps import get_app, get_app_deployments, get_apps, update_app_resources from ..db.audit import log_audit_event from ..db.jobs import create_job, has_active_deploy_job from ..routes.deployments import render_status_pill from ..shell import run_command from ..templates.layout import page, render_result router = APIRouter() @router.get("/", response_class=HTMLResponse) def index(request: Request, user=Depends(require_user)): apps = get_apps() gitea_url = read_env_value("GITEA_URL", "") gitea_org = read_env_value("GITEA_ORG", DEFAULT_GITEA_ORG) host = read_env_value("APPFACTORY_HOST", DEFAULT_APPFACTORY_HOST) rows = "" for item in apps: app_id = html.escape(item.get("id", "")) app_url_id = quote(item.get("id", ""), safe="") status = html.escape(item.get("status", "")) docs = html.escape(item.get("docs", f"/apps/{app_id}/docs")) memory = item.get("memory", "") cpus = item.get("cpus", "") http_clone = html.escape(f"git clone {gitea_url}/{gitea_org}/{app_id}.git") ssh_clone = html.escape(f"git clone ssh://git@{host}:2222/{gitea_org}/{app_id}.git") memory_options = "" for value, label in [ ("", "Výchozí"), ("256m", "256 MB - malá služba"), ("512m", "512 MB - běžná služba"), ("1g", "1 GB - větší služba"), ("2g", "2 GB - náročná služba"), ]: selected = "selected" if memory == value else "" memory_options += f'' cpu_options = "" for value, label in [ ("", "Výchozí"), ("0.25", "0,25 CPU - velmi malá služba"), ("0.50", "0,50 CPU - běžná služba"), ("1.00", "1 CPU - celé jádro"), ("2.00", "2 CPU - náročná služba"), ]: selected = "selected" if cpus == value else "" cpu_options += f'' rows += f""" {app_id}
/apps/{app_id} {status} Swagger
Paměť je limit RAM. CPU určuje maximální podíl výpočetního výkonu. Příklad: 0,50 = polovina jádra, 1,00 = celé jádro.
Příkazy pro klonování

Detail

Nasazení

""" if not rows: rows = 'Zatím nejsou nasazené žádné aplikace.' return page( "Aplikace", f"""

Aplikace

Vytváření, nasazení, klonování, nastavení prostředků a mazání služeb.

+ Nová aplikace

Zálohy

Vytváření záloh a kopírování příkazů pro obnovu. Obnova je záměrně ruční a chráněná.

Spravovat zálohy

Nasazení

Historie posledních běhů nasazení, stavů a výstupů z deploy procesu.

Zobrazit nasazení

Nasazené aplikace

{rows}
Aplikace Status Dokumentace Prostředky Git Akce
""", user=user, ) @router.get("/apps/{app_id}", response_class=HTMLResponse) def app_detail(app_id: str, request: Request, user=Depends(require_user)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") escaped_app_id = html.escape(app.get("id", "")) app_url_id = quote(app.get("id", ""), safe="") name = html.escape(app.get("name", "") or "") language = html.escape(app.get("language", "") or "") version = html.escape(app.get("version", "") or "") status = html.escape(app.get("status", "") or "") memory = html.escape(app.get("memory", "") or "") cpus = html.escape(app.get("cpus", "") or "") updated_at = html.escape(app.get("updated_at", "") or "") rows = "" for deployment in get_app_deployments(app.get("id", ""), limit=10): deployment_id = html.escape(str(deployment.get("id", ""))) started_at = html.escape(deployment.get("started_at", "") or "") triggered_by = html.escape( deployment.get("triggered_by_display_name") or deployment.get("triggered_by_username") or "" ) rows += f""" #{deployment_id} {render_status_pill(deployment.get("status"))} {started_at} {triggered_by} """ if not rows: rows = 'Zatím nejsou evidovaná žádná nasazení této aplikace.' return page( f"Aplikace {escaped_app_id}", f"""

{escaped_app_id}

{name}

← Zpět na aplikace Nasazení aplikace

Souhrn

ID{escaped_app_id}
Název{name}
Jazyk{language}
Verze{version}
Status{status}
Paměť{memory}
CPU{cpus}
Upraveno{updated_at}

Poslední deploymenty

{rows}
ID Status Timestamp Triggered by
""", user=user, ) @router.post("/apps/{app_id}/redeploy") def redeploy_app(app_id: str, user=Depends(require_user)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") active_job = has_active_deploy_job("app", app_id) if active_job: return RedirectResponse(url=f"/portal/jobs/{active_job['id']}", status_code=303) job_id = create_job( job_type="deploy_app", target_type="app", target_id=app_id, payload={}, user=user, source="portal", ) log_audit_event( user, action="app.redeploy.queued", target_type="app", target_id=app_id, metadata={"job_id": job_id}, ) return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303) @router.get("/new-app", response_class=HTMLResponse) def new_app_form(request: Request, user=Depends(require_user)): return page( "Nová aplikace", """

Vytvořit novou aplikaci

Vytvoří Gitea repozitář, webhook, lokální workspace, první commit a nasadí službu.




← Zpět

""", user=user, ) @router.post("/new-app", response_class=HTMLResponse) def create_app( app_id: str = Form(...), app_name: str = Form(...), template: str = Form(...), user=Depends(require_user), ): if template != "python-fastapi": return HTMLResponse("Nepodporovaná šablona", status_code=400) create_result = run_command([NEW_APP_SCRIPT, app_id, app_name]) deploy_result = run_command([DEPLOY_SCRIPT, app_id]) status = "OK" if create_result.returncode == 0 and deploy_result.returncode == 0 else "FAILED" log_audit_event( user, action="create_app", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "app_name": app_name, "template": template, "status": status, "create_returncode": create_result.returncode, "deploy_returncode": deploy_result.returncode, }, ) return render_result( title=f"Vytvoření aplikace: {status}", back_url="/portal", sections=[ ("Výstup vytvoření", create_result.stdout), ("Chyba vytvoření", create_result.stderr), ("Výstup nasazení", deploy_result.stdout), ("Chyba nasazení", deploy_result.stderr), ], extra_link=f"/apps/{html.escape(app_id)}/docs", extra_label="Otevřít Swagger", user=user, ) @router.post("/delete-app", response_class=HTMLResponse) def delete_app(app_id: str = Form(...), user=Depends(require_user)): result = run_command([DELETE_APP_SCRIPT, app_id]) status = "OK" if result.returncode == 0 else "FAILED" log_audit_event( user, action="delete_app", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "status": status, "returncode": result.returncode, }, ) return render_result( title=f"Smazání aplikace: {status}", back_url="/portal", sections=[("Výstup", result.stdout), ("Chyba", result.stderr)], user=user, ) @router.post("/update-resources", response_class=HTMLResponse) def update_resources( app_id: str = Form(...), memory: str = Form(""), cpus: str = Form(""), user=Depends(require_user), ): memory = memory.strip() cpus = cpus.strip() update_app_resources(app_id, memory, cpus) catalog_result = run_command(["/tools/generate-catalog.sh"]) compose_result = run_command([GENERATE_COMPOSE_SCRIPT]) deploy_result = run_command([DEPLOY_SCRIPT, app_id]) status = ( "OK" if catalog_result.returncode == 0 and compose_result.returncode == 0 and deploy_result.returncode == 0 else "FAILED" ) log_audit_event( user, action="update_resources", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "memory": memory, "cpus": cpus, "status": status, "catalog_returncode": catalog_result.returncode, "compose_returncode": compose_result.returncode, "deploy_returncode": deploy_result.returncode, }, ) return render_result( title=f"Úprava prostředků: {status}", back_url="/portal", sections=[ ("Výstup katalogu", catalog_result.stdout), ("Chyba katalogu", catalog_result.stderr), ("Výstup compose", compose_result.stdout), ("Chyba compose", compose_result.stderr), ("Výstup nasazení", deploy_result.stdout), ("Chyba nasazení", deploy_result.stderr), ], user=user, )