refactor: FastAPI best practices — return types, Pydantic schemas, middleware, code dedup
- Все 18 роутов получили return type annotations
- Создан schemas.py с Pydantic-моделями (ScanOut, PackageOut, FindingOut, ...)
- API-роуты: response_model на list/detail/export/stats
- 404 через HTTPException(404) вместо {'detail':'Not found'} (200)
- RequestLoggingMiddleware: method, path, status, duration_ms
- Глобальный exception handler: ловит необработанные исключения → 500
- _parse_flagged(): вынесен дублирующийся string→bool
- parse_package_path(): общий для web.py и api_packages.py
- selectinload: вынесены в top-level imports
- harvester: makedirs/mkdtemp/rmtree обёрнуты в asyncio.to_thread()
This commit is contained in:
@@ -1,13 +1,13 @@
|
||||
"""Web UI routes — Jinja2 + htmx pages."""
|
||||
|
||||
import asyncio
|
||||
from urllib.parse import unquote
|
||||
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
from fastapi.responses import HTMLResponse
|
||||
from jinja2 import Environment, PackageLoader, select_autoescape
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from ..config import config
|
||||
from ..constants import (
|
||||
@@ -17,6 +17,7 @@ from ..constants import (
|
||||
DEFAULT_SORT_DIR,
|
||||
WEB_PER_PAGE,
|
||||
)
|
||||
from ..core.nexus import parse_package_path
|
||||
from ..db.engine import get_session
|
||||
from ..db.models import Finding, Scan
|
||||
from ..db.queries import (
|
||||
@@ -45,14 +46,20 @@ def _render(name: str, **context) -> HTMLResponse:
|
||||
return HTMLResponse(template.render(**context), status_code=status_code)
|
||||
|
||||
|
||||
def _parse_flagged(value: str) -> bool | None:
|
||||
return True if value == "1" else None
|
||||
|
||||
|
||||
@router.get("/", response_class=HTMLResponse)
|
||||
async def dashboard(request: Request, session: AsyncSession = Depends(get_session)):
|
||||
async def dashboard(request: Request, session: AsyncSession = Depends(get_session)) -> HTMLResponse:
|
||||
ctx = await get_dashboard_stats(session)
|
||||
return _render("dashboard.html", **ctx, request=request)
|
||||
|
||||
|
||||
@router.get("/dashboard/stats", response_class=HTMLResponse)
|
||||
async def dashboard_stats_fragment(request: Request, session: AsyncSession = Depends(get_session)):
|
||||
async def dashboard_stats_fragment(
|
||||
request: Request, session: AsyncSession = Depends(get_session)
|
||||
) -> HTMLResponse:
|
||||
ctx = await get_dashboard_stats(session)
|
||||
return _render("dashboard_stats.html", request=request, **ctx)
|
||||
|
||||
@@ -67,13 +74,11 @@ async def scans_list(
|
||||
sort_by: str = DEFAULT_SORT_BY_SCANS,
|
||||
sort_dir: str = DEFAULT_SORT_DIR,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
) -> HTMLResponse:
|
||||
per_page = WEB_PER_PAGE
|
||||
offset = (page - 1) * per_page
|
||||
|
||||
flagged_bool = None
|
||||
if flagged == "1":
|
||||
flagged_bool = True
|
||||
flagged_bool = _parse_flagged(flagged)
|
||||
|
||||
q, count_q = build_scan_list_query(
|
||||
flagged=flagged_bool,
|
||||
@@ -105,9 +110,9 @@ async def scans_list(
|
||||
|
||||
|
||||
@router.get("/scans/{scan_id}", response_class=HTMLResponse)
|
||||
async def scan_detail(scan_id: int, request: Request, session: AsyncSession = Depends(get_session)):
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
async def scan_detail(
|
||||
scan_id: int, request: Request, session: AsyncSession = Depends(get_session)
|
||||
) -> HTMLResponse:
|
||||
scan = await session.scalar(
|
||||
select(Scan).where(Scan.id == scan_id).options(selectinload(Scan.findings))
|
||||
)
|
||||
@@ -126,13 +131,11 @@ async def packages_list(
|
||||
sort_by: str = DEFAULT_SORT_BY_PACKAGES,
|
||||
sort_dir: str = DEFAULT_SORT_DIR,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
) -> HTMLResponse:
|
||||
per_page = WEB_PER_PAGE
|
||||
offset = (page - 1) * per_page
|
||||
|
||||
flagged_bool = None
|
||||
if flagged == "1":
|
||||
flagged_bool = True
|
||||
flagged_bool = _parse_flagged(flagged)
|
||||
|
||||
rows_q, total_q = build_package_list_query(
|
||||
flagged=flagged_bool,
|
||||
@@ -166,14 +169,8 @@ async def package_detail(
|
||||
name: str,
|
||||
request: Request,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
# name:path captures the entire path after /packages/
|
||||
# e.g. "eviltest/0.1.0" or "github.com/attacker/evilmodule/v0.1.0"
|
||||
parts = name.rsplit("/", 1)
|
||||
pkg_name = unquote(parts[0])
|
||||
pkg_version = unquote(parts[1]) if len(parts) == 2 else ""
|
||||
|
||||
from sqlalchemy.orm import selectinload
|
||||
) -> HTMLResponse:
|
||||
pkg_name, pkg_version = parse_package_path(name)
|
||||
|
||||
scans = (
|
||||
(
|
||||
@@ -211,7 +208,7 @@ async def analyze_finding_htmx(
|
||||
request: Request,
|
||||
retry: bool = False,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
) -> HTMLResponse:
|
||||
"""HTMX fragment: trigger LLM analysis and return styled result HTML."""
|
||||
from ..config import config
|
||||
from ..core.llm import analyze_finding
|
||||
|
||||
Reference in New Issue
Block a user