Merge pull request #1385 from govfvck/doc-extra-responses

remove fastapi_responses, add manual additional responses docs

app/__init__.py

@@ -1,29 +1,29 @@
 import logging
+
 from apscheduler.schedulers.background import BackgroundScheduler
 from fastapi import FastAPI, Request, status
 from fastapi.encoders import jsonable_encoder
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse, HTMLResponse
+from fastapi.responses import JSONResponse
 from fastapi.routing import APIRoute
-from fastapi_responses import custom_openapi
-
-from config import DOCS, XRAY_SUBSCRIPTION_PATH, ALLOWED_ORIGINS
 
+from config import ALLOWED_ORIGINS, DOCS, XRAY_SUBSCRIPTION_PATH
 
 __version__ = "0.7.0"
 
 app = FastAPI(
     title="MarzbanAPI",
     description="Unified GUI Censorship Resistant Solution Powered by Xray",
     version=__version__,
-    docs_url='/docs' if DOCS else None,
-    redoc_url='/redoc' if DOCS else None
+    docs_url="/docs" if DOCS else None,
+    redoc_url="/redoc" if DOCS else None,
 )
 
-app.openapi = custom_openapi(app)
-scheduler = BackgroundScheduler({'apscheduler.job_defaults.max_instances': 20}, timezone='UTC')
-logger = logging.getLogger('uvicorn.error')
+scheduler = BackgroundScheduler(
+    {"apscheduler.job_defaults.max_instances": 20}, timezone="UTC"
+)
+logger = logging.getLogger("uvicorn.error")
 
 app.add_middleware(
     CORSMiddleware,
@@ -53,7 +53,9 @@ def on_startup():
     paths = [f"{r.path}/" for r in app.routes]
     paths.append("/api/")
     if f"/{XRAY_SUBSCRIPTION_PATH}/" in paths:
-        raise ValueError(f"you can't use /{XRAY_SUBSCRIPTION_PATH}/ as subscription path it reserved for {app.title}")
+        raise ValueError(
+            f"you can't use /{XRAY_SUBSCRIPTION_PATH}/ as subscription path it reserved for {app.title}"
+        )
     scheduler.start()
 
 

app/routers/admin.py

@@ -1,16 +1,17 @@
 from typing import List, Optional
 
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.exc import IntegrityError
+
 from app.db import Session, crud, get_db
+from app.dependencies import get_admin_by_username, validate_admin
 from app.models.admin import Admin, AdminCreate, AdminModify, Token
+from app.utils import report, responses
 from app.utils.jwt import create_admin_token
-from fastapi import Depends, HTTPException, status, Request, APIRouter
-from fastapi.security import OAuth2PasswordRequestForm
-from app.utils import report
-from app.dependencies import validate_admin, get_admin_by_username
 from config import LOGIN_NOTIFY_WHITE_LIST
 
-router = APIRouter(tags=['Admin'], prefix='/api')
+router = APIRouter(tags=["Admin"], prefix="/api")
 
 
 def get_client_ip(request: Request) -> str:
@@ -27,7 +28,7 @@ def get_client_ip(request: Request) -> str:
 def admin_token(
     request: Request,
     form_data: OAuth2PasswordRequestForm = Depends(),
-    db: Session = Depends(get_db)
+    db: Session = Depends(get_db),
 ):
     """Authenticate an admin and issue a token."""
     client_ip = get_client_ip(request)
@@ -42,21 +43,20 @@ def admin_token(
         )
 
     if client_ip not in LOGIN_NOTIFY_WHITE_LIST:
-        report.login(form_data.username, '🔒', client_ip, True)
-
-    return Token(
-        access_token=create_admin_token(
-            form_data.username,
-            dbadmin.is_sudo
-        )
-    )
+        report.login(form_data.username, "🔒", client_ip, True)
 
+    return Token(access_token=create_admin_token(form_data.username, dbadmin.is_sudo))
 
-@router.post("/admin", response_model=Admin)
+
+@router.post(
+    "/admin",
+    response_model=Admin,
+    responses={401: responses._401, 403: responses._403, 409: responses._409},
+)
 def create_admin(
     new_admin: AdminCreate,
     db: Session = Depends(get_db),
-    admin: Admin = Depends(Admin.check_sudo_admin)
+    admin: Admin = Depends(Admin.check_sudo_admin),
 ):
     """Create a new admin if the current admin has sudo privileges."""
     try:
@@ -68,58 +68,66 @@ def create_admin(
     return dbadmin
 
 
-@router.put("/admin/{username}", response_model=Admin)
+@router.put(
+    "/admin/{username}",
+    response_model=Admin,
+    responses={401: responses._401, 403: responses._403},
+)
 def modify_admin(
     modified_admin: AdminModify,
     dbadmin: Admin = Depends(get_admin_by_username),
     db: Session = Depends(get_db),
-    current_admin: Admin = Depends(Admin.check_sudo_admin)
+    current_admin: Admin = Depends(Admin.check_sudo_admin),
 ):
     """Modify an existing admin's details."""
     if (dbadmin.username != current_admin.username) and dbadmin.is_sudo:
         raise HTTPException(
             status_code=403,
-            detail="You're not allowed to edit another sudoer's account. Use marzban-cli instead."
+            detail="You're not allowed to edit another sudoer's account. Use marzban-cli instead.",
         )
-    
+
     updated_admin = crud.update_admin(db, dbadmin, modified_admin)
-    
+
     return updated_admin
 
 
-@router.delete("/admin/{username}")
+@router.delete(
+    "/admin/{username}",
+    responses={401: responses._401, 403: responses._403},
+)
 def remove_admin(
     dbadmin: Admin = Depends(get_admin_by_username),
     db: Session = Depends(get_db),
-    current_admin: Admin = Depends(Admin.check_sudo_admin)
+    current_admin: Admin = Depends(Admin.check_sudo_admin),
 ):
     """Remove an admin from the database."""
     if dbadmin.is_sudo:
         raise HTTPException(
             status_code=403,
-            detail="You're not allowed to delete sudo accounts. Use marzban-cli instead."
+            detail="You're not allowed to delete sudo accounts. Use marzban-cli instead.",
         )
 
     crud.remove_admin(db, dbadmin)
-
     return {"detail": "Admin removed successfully"}
 
 
-@router.get("/admin", response_model=Admin)
-def get_current_admin(
-    admin: Admin = Depends(Admin.get_current)
-):
+@router.get("/admin", response_model=Admin, responses={401: responses._401})
+def get_current_admin(admin: Admin = Depends(Admin.get_current)):
     """Retrieve the current authenticated admin."""
     return admin
 
 
-@router.get("/admins", response_model=List[Admin])
+@router.get(
+    "/admins",
+    response_model=List[Admin],
+    responses={401: responses._401, 403: responses._403},
+)
 def get_admins(
     offset: Optional[int] = None,
     limit: Optional[int] = None,
     username: Optional[str] = None,
     db: Session = Depends(get_db),
-    admin: Admin = Depends(Admin.check_sudo_admin)
+    admin: Admin = Depends(Admin.check_sudo_admin),
 ):
     """Fetch a list of admins with optional filters for pagination and username."""
-    return crud.get_admins(db, offset, limit, username)
+    return crud.get_admins(db, offset, limit, username)

app/routers/core.py

@@ -3,44 +3,46 @@
 import time
 
 import commentjson
-from fastapi import Depends, HTTPException, WebSocket, APIRouter
+from fastapi import APIRouter, Depends, HTTPException, WebSocket
 from starlette.websockets import WebSocketDisconnect
 
 from app import xray
 from app.db import Session, get_db
 from app.models.admin import Admin
 from app.models.core import CoreStats
+from app.utils import responses
 from app.xray import XRayConfig
 from config import XRAY_JSON
 
-router = APIRouter(tags=['Core'], prefix='/api')
+router = APIRouter(tags=["Core"], prefix="/api", responses={401: responses._401})
+
 
 @router.websocket("/core/logs")
 async def core_logs(websocket: WebSocket, db: Session = Depends(get_db)):
-    token = (
-        websocket.query_params.get('token')
-        or
-        websocket.headers.get('Authorization', '').removeprefix("Bearer ")
-    )
+    token = websocket.query_params.get("token") or websocket.headers.get(
+        "Authorization", ""
+    ).removeprefix("Bearer ")
     admin = Admin.get_admin(token, db)
     if not admin:
         return await websocket.close(reason="Unauthorized", code=4401)
 
     if not admin.is_sudo:
         return await websocket.close(reason="You're not allowed", code=4403)
 
-    interval = websocket.query_params.get('interval')
+    interval = websocket.query_params.get("interval")
     if interval:
         try:
             interval = float(interval)
         except ValueError:
             return await websocket.close(reason="Invalid interval value", code=4400)
         if interval > 10:
-            return await websocket.close(reason="Interval must be more than 0 and at most 10 seconds", code=4400)
+            return await websocket.close(
+                reason="Interval must be more than 0 and at most 10 seconds", code=4400
+            )
 
     await websocket.accept()
 
-    cache = ''
+    cache = ""
     last_sent_ts = 0
     with xray.core.get_logs() as logs:
         while True:
@@ -49,7 +51,7 @@ async def core_logs(websocket: WebSocket, db: Session = Depends(get_db)):
                     await websocket.send_text(cache)
                 except (WebSocketDisconnect, RuntimeError):
                     break
-                cache = ''
+                cache = ""
                 last_sent_ts = time.time()
 
             if not logs:
@@ -64,50 +66,48 @@ async def core_logs(websocket: WebSocket, db: Session = Depends(get_db)):
             log = logs.popleft()
 
             if interval:
-                cache += f'{log}\n'
+                cache += f"{log}\n"
                 continue
 
             try:
                 await websocket.send_text(log)
             except (WebSocketDisconnect, RuntimeError):
                 break
 
+
 @router.get("/core", response_model=CoreStats)
-def get_core_stats(
-    admin: Admin = Depends(Admin.get_current)
-):
+def get_core_stats(admin: Admin = Depends(Admin.get_current)):
     """Retrieve core statistics such as version and uptime."""
     return CoreStats(
         version=xray.core.version,
         started=xray.core.started,
-        logs_websocket=router.url_path_for('core_logs')
+        logs_websocket=router.url_path_for("core_logs"),
     )
 
-@router.post("/core/restart")
-def restart_core(
-    admin: Admin = Depends(Admin.check_sudo_admin)
-):
+
+@router.post("/core/restart", responses={403: responses._403})
+def restart_core(admin: Admin = Depends(Admin.check_sudo_admin)):
     """Restart the core and all connected nodes."""
     startup_config = xray.config.include_db_users()
     xray.core.restart(startup_config)
 
     for node_id, node in list(xray.nodes.items()):
         if node.connected:
             xray.operations.restart_node(node_id, startup_config)
-    
+
     return {}
 
-@router.get("/core/config")
-def get_core_config(
-    admin: Admin = Depends(Admin.check_sudo_admin)
-) -> dict:
+
+@router.get("/core/config", responses={403: responses._403})
+def get_core_config(admin: Admin = Depends(Admin.check_sudo_admin)) -> dict:
     """Get the current core configuration."""
     with open(XRAY_JSON, "r") as f:
         config = commentjson.loads(f.read())
 
     return config
 
-@router.put("/core/config")
+
+@router.put("/core/config", responses={403: responses._403})
 def modify_core_config(
     payload: dict, admin: Admin = Depends(Admin.check_sudo_admin)
 ) -> dict:
@@ -118,7 +118,7 @@ def modify_core_config(
         raise HTTPException(status_code=400, detail=str(err))
 
     xray.config = config
-    with open(XRAY_JSON, 'w') as f:
+    with open(XRAY_JSON, "w") as f:
         f.write(json.dumps(payload, indent=4))
 
     startup_config = xray.config.include_db_users()