bstein-dev-home/backend/atlas_portal/routes/admin_access.py

from __future__ import annotations

from typing import Any

from flask import jsonify, g

from ..db import connect, configured
from ..keycloak import require_auth, require_portal_admin


def register(app) -> None:
    @app.route("/api/admin/access/requests", methods=["GET"])
    @require_auth
    def admin_list_requests() -> Any:
        ok, resp = require_portal_admin()
        if not ok:
            return resp
        if not configured():
            return jsonify({"error": "server not configured"}), 503

        try:
            with connect() as conn:
                rows = conn.execute(
                    """
                    SELECT request_code, username, contact_email, note, status, created_at
                    FROM access_requests
                    WHERE status = 'pending'
                    ORDER BY created_at ASC
                    LIMIT 200
                    """
                ).fetchall()
        except Exception:
            return jsonify({"error": "failed to load requests"}), 502

        output: list[dict[str, Any]] = []
        for row in rows:
            output.append(
                {
                    "id": row["request_code"],
                    "username": row["username"],
                    "email": row.get("contact_email") or "",
                    "request_code": row["request_code"],
                    "created_at": (row.get("created_at").isoformat() if row.get("created_at") else ""),
                    "note": row.get("note") or "",
                }
            )
        return jsonify({"requests": output})

    @app.route("/api/admin/access/requests/<username>/approve", methods=["POST"])
    @require_auth
    def admin_approve_request(username: str) -> Any:
        ok, resp = require_portal_admin()
        if not ok:
            return resp
        if not configured():
            return jsonify({"error": "server not configured"}), 503

        decided_by = getattr(g, "keycloak_username", "") or ""
        try:
            with connect() as conn:
                row = conn.execute(
                    """
                    UPDATE access_requests
                    SET status = 'approved', decided_at = NOW(), decided_by = %s
                    WHERE username = %s AND status = 'pending'
                    RETURNING request_code
                    """,
                    (decided_by or None, username),
                ).fetchone()
        except Exception:
            return jsonify({"error": "failed to approve request"}), 502

        if not row:
            return jsonify({"ok": True, "request_code": ""})
        return jsonify({"ok": True, "request_code": row["request_code"]})

    @app.route("/api/admin/access/requests/<username>/deny", methods=["POST"])
    @require_auth
    def admin_deny_request(username: str) -> Any:
        ok, resp = require_portal_admin()
        if not ok:
            return resp
        if not configured():
            return jsonify({"error": "server not configured"}), 503

        decided_by = getattr(g, "keycloak_username", "") or ""
        try:
            with connect() as conn:
                row = conn.execute(
                    """
                    UPDATE access_requests
                    SET status = 'denied', decided_at = NOW(), decided_by = %s
                    WHERE username = %s AND status = 'pending'
                    RETURNING request_code
                    """,
                    (decided_by or None, username),
                ).fetchone()
        except Exception:
            return jsonify({"error": "failed to deny request"}), 502

        if not row:
            return jsonify({"ok": True, "request_code": ""})
        return jsonify({"ok": True, "request_code": row["request_code"]})