feat: implement in-memory caching for dashboard data and background warmup task

This commit is contained in:
Lorenzo Venerandi
2026-03-10 11:00:22 +01:00
parent 56345974c8
commit 1eb4f54f5c
5 changed files with 145 additions and 21 deletions

View File

@@ -18,6 +18,7 @@ from pydantic import BaseModel
from dependencies import get_db, get_client_ip
from logger import get_app_logger
from dashboard_cache import get_cached, is_warm
# Server-side session token store (valid tokens for authenticated sessions)
_auth_tokens: set = set()
@@ -249,10 +250,16 @@ async def all_ips(
sort_by: str = Query("total_requests"),
sort_order: str = Query("desc"),
):
db = get_db()
page = max(1, page)
page_size = min(max(1, page_size), 10000)
# Serve from cache on default map request (top 100 IPs)
if page == 1 and page_size == 100 and sort_by == "total_requests" and sort_order == "desc" and is_warm():
cached = get_cached("map_ips")
if cached:
return JSONResponse(content=cached, headers=_no_cache_headers())
db = get_db()
try:
result = db.get_all_ips_paginated(
page=page, page_size=page_size, sort_by=sort_by, sort_order=sort_order

View File

@@ -10,6 +10,7 @@ from fastapi.responses import JSONResponse
from logger import get_app_logger
from dependencies import get_db, get_templates
from dashboard_cache import get_cached, is_warm
router = APIRouter()
@@ -17,17 +18,19 @@ router = APIRouter()
@router.get("")
@router.get("/")
async def dashboard_page(request: Request):
db = get_db()
config = request.app.state.config
dashboard_path = "/" + config.dashboard_secret_path.lstrip("/")
# Get initial data for server-rendered sections
stats = db.get_dashboard_counts()
suspicious = db.get_recent_suspicious(limit=10)
# Get credential count for the stats card
cred_result = db.get_credentials_paginated(page=1, page_size=1)
stats["credential_count"] = cred_result["pagination"]["total"]
# Serve from pre-computed cache when available, fall back to live queries
if is_warm():
stats = get_cached("stats")
suspicious = get_cached("suspicious")
else:
db = get_db()
stats = db.get_dashboard_counts()
suspicious = db.get_recent_suspicious(limit=10)
cred_result = db.get_credentials_paginated(page=1, page_size=1)
stats["credential_count"] = cred_result["pagination"]["total"]
templates = get_templates()
return templates.TemplateResponse(

View File

@@ -10,6 +10,7 @@ from fastapi.responses import HTMLResponse
from dependencies import get_db, get_templates
from routes.api import verify_auth
from dashboard_cache import get_cached, is_warm
router = APIRouter()
@@ -58,10 +59,15 @@ async def htmx_top_ips(
sort_by: str = Query("count"),
sort_order: str = Query("desc"),
):
db = get_db()
result = db.get_top_ips_paginated(
page=max(1, page), page_size=8, sort_by=sort_by, sort_order=sort_order
)
# Serve from cache on default first-page request
cached = get_cached("top_ips") if (page == 1 and sort_by == "count" and sort_order == "desc" and is_warm()) else None
if cached:
result = cached
else:
db = get_db()
result = db.get_top_ips_paginated(
page=max(1, page), page_size=8, sort_by=sort_by, sort_order=sort_order
)
templates = get_templates()
return templates.TemplateResponse(
@@ -87,10 +93,14 @@ async def htmx_top_paths(
sort_by: str = Query("count"),
sort_order: str = Query("desc"),
):
db = get_db()
result = db.get_top_paths_paginated(
page=max(1, page), page_size=5, sort_by=sort_by, sort_order=sort_order
)
cached = get_cached("top_paths") if (page == 1 and sort_by == "count" and sort_order == "desc" and is_warm()) else None
if cached:
result = cached
else:
db = get_db()
result = db.get_top_paths_paginated(
page=max(1, page), page_size=5, sort_by=sort_by, sort_order=sort_order
)
templates = get_templates()
return templates.TemplateResponse(
@@ -116,10 +126,14 @@ async def htmx_top_ua(
sort_by: str = Query("count"),
sort_order: str = Query("desc"),
):
db = get_db()
result = db.get_top_user_agents_paginated(
page=max(1, page), page_size=5, sort_by=sort_by, sort_order=sort_order
)
cached = get_cached("top_ua") if (page == 1 and sort_by == "count" and sort_order == "desc" and is_warm()) else None
if cached:
result = cached
else:
db = get_db()
result = db.get_top_user_agents_paginated(
page=max(1, page), page_size=5, sort_by=sort_by, sort_order=sort_order
)
templates = get_templates()
return templates.TemplateResponse(