Add API lab

2026-06-02 20:41:41 +02:00
parent 3f0a5ecf9b
commit 3997b695f3
5 changed files with 229 additions and 0 deletions
@@ -0,0 +1,206 @@
+import json
+from typing import Any
+
+import requests
+import streamlit as st
+
+
+st.set_page_config(page_title="Visione Lab", layout="wide")
+
+
+def api_url(path: str) -> str:
+    return st.session_state.base_url.rstrip("/") + path
+
+
+def headers() -> dict[str, str]:
+    token = st.session_state.get("token", "")
+    return {"Authorization": f"Bearer {token}"} if token else {}
+
+
+def show_response(response: requests.Response) -> None:
+    st.caption(f"{response.request.method} {response.url} -> {response.status_code}")
+    try:
+        st.json(response.json())
+    except ValueError:
+        st.text(response.text)
+
+
+def json_input(label: str, value: dict[str, Any]) -> dict[str, Any] | None:
+    raw = st.text_area(label, json.dumps(value, indent=2), height=180)
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError as error:
+        st.error(f"Invalid JSON: {error}")
+        return None
+
+
+if "base_url" not in st.session_state:
+    st.session_state.base_url = "http://127.0.0.1:5555/api/v1"
+
+with st.sidebar:
+    st.text_input("Backend API", key="base_url")
+    st.text_input("Bearer token", key="token", type="password")
+    st.link_button("MailHog", "http://127.0.0.1:8025")
+    if st.button("Health check"):
+        show_response(requests.get(api_url("/health"), timeout=5))
+
+st.title("Visione API Lab")
+
+auth_tab, reports_tab, admin_tab, docs_tab = st.tabs(
+    ["Auth", "Reports", "Admin", "Docs"]
+)
+
+with auth_tab:
+    col1, col2 = st.columns(2)
+    with col1:
+        st.subheader("Password login")
+        email = st.text_input("Email", "[email protected]")
+        password = st.text_input("Password", "ChangeMe123!", type="password")
+        if st.button("Login"):
+            response = requests.post(
+                api_url("/auth/login/password"),
+                json={"email": email, "password": password},
+                timeout=10,
+            )
+            data = response.json() if response.headers.get("content-type", "").startswith("application/json") else {}
+            token = data.get("token")
+            if token:
+                st.session_state.token = token
+            show_response(response)
+
+        st.subheader("OTP")
+        otp_email = st.text_input("OTP email", "[email protected]")
+        if st.button("Request OTP"):
+            show_response(
+                requests.post(
+                    api_url("/auth/otp/request"),
+                    json={"email": otp_email},
+                    timeout=10,
+                )
+            )
+        otp_code = st.text_input("OTP code")
+        if st.button("Verify OTP"):
+            response = requests.post(
+                api_url("/auth/otp/verify"),
+                json={"email": otp_email, "code": otp_code},
+                timeout=10,
+            )
+            data = response.json() if response.headers.get("content-type", "").startswith("application/json") else {}
+            token = data.get("token")
+            if token:
+                st.session_state.token = token
+            show_response(response)
+
+    with col2:
+        st.subheader("Register citizen")
+        payload = json_input(
+            "Register payload",
+            {
+                "first_name": "Mario",
+                "last_name": "Rossi",
+                "email": "[email protected]",
+                "password": "ChangeMe123!",
+                "fiscal_code": "RSSMRA80A01H501U",
+                "phone": "+391234567890",
+            },
+        )
+        if st.button("Register") and payload:
+            show_response(requests.post(api_url("/auth/register"), json=payload, timeout=10))
+
+        st.subheader("Me")
+        if st.button("GET /me"):
+            show_response(requests.get(api_url("/me"), headers=headers(), timeout=10))
+
+with reports_tab:
+    col1, col2 = st.columns(2)
+    with col1:
+        st.subheader("Create report")
+        report_payload = json_input(
+            "Report payload",
+            {
+                "category_id": 1,
+                "latitude": 41.9028,
+                "longitude": 12.4964,
+                "address": "Via Roma 1",
+                "description": "Lampione non funzionante",
+            },
+        )
+        if st.button("Create report") and report_payload:
+            show_response(
+                requests.post(
+                    api_url("/reports"),
+                    json=report_payload,
+                    headers=headers(),
+                    timeout=10,
+                )
+            )
+
+        st.subheader("List reports")
+        status = st.text_input("Status filter")
+        params = {"status": status} if status else {}
+        if st.button("GET /reports"):
+            show_response(
+                requests.get(
+                    api_url("/reports"),
+                    params=params,
+                    headers=headers(),
+                    timeout=10,
+                )
+            )
+
+    with col2:
+        report_id = st.number_input("Report id", min_value=1, step=1)
+        if st.button("Detail"):
+            show_response(
+                requests.get(
+                    api_url(f"/reports/{report_id}"),
+                    headers=headers(),
+                    timeout=10,
+                )
+            )
+
+        st.subheader("Upload media")
+        files = st.file_uploader(
+            "Images or videos",
+            accept_multiple_files=True,
+            type=["jpg", "jpeg", "png", "webp", "mp4", "webm", "mov"],
+        )
+        if st.button("Upload") and files:
+            multipart = [
+                ("file", (file.name, file.getvalue(), file.type or "application/octet-stream"))
+                for file in files
+            ]
+            show_response(
+                requests.post(
+                    api_url(f"/reports/{report_id}/media"),
+                    headers=headers(),
+                    files=multipart,
+                    timeout=60,
+                )
+            )
+
+        st.subheader("Workflow")
+        new_status = st.selectbox(
+            "Status",
+            ["UNDER_REVIEW", "ASSIGNED", "IN_PROGRESS", "RESOLVED", "REJECTED", "CLOSED"],
+        )
+        if st.button("Update status"):
+            show_response(
+                requests.patch(
+                    api_url(f"/reports/{report_id}/status"),
+                    json={"status": new_status, "note": "Updated from Visione Lab"},
+                    headers=headers(),
+                    timeout=10,
+                )
+            )
+
+with admin_tab:
+    st.subheader("Reference data")
+    endpoint = st.selectbox("Endpoint", ["/admin/categories", "/admin/areas", "/admin/teams", "/admin/stats/overview", "/admin/users"])
+    if st.button("GET admin endpoint"):
+        show_response(requests.get(api_url(endpoint), headers=headers(), timeout=10))
+
+with docs_tab:
+    st.link_button("Swagger UI", api_url("/docs"))
+    if st.button("Fetch OpenAPI"):
+        show_response(requests.get(api_url("/openapi.json"), timeout=10))
@@ -0,0 +1,2 @@
+streamlit>=1.35
+requests>=2.31