Skip to content

Dashboard

Dashboard Statistics API Endpoint.

Provides REST endpoint for dashboard stats and manages periodic WebSocket broadcasts.

broadcast_dashboard_stats_loop(interval_seconds=5) async

Background task that periodically broadcasts dashboard stats via WebSocket.

Parameters:

  • interval_seconds (int, default: 5 ) –

    Broadcast interval in seconds (default: 5).

Returns:

  • None –

    None

Source code in app/api/v1/endpoints/dashboard.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def broadcast_dashboard_stats_loop(interval_seconds: int = 5) -> None:
    """
    Background task that periodically broadcasts dashboard stats via WebSocket.

    Args:
        interval_seconds: Broadcast interval in seconds (default: 5).

    Returns:
        None
    """
    global _broadcast_running
    _broadcast_running = True

    logger.info("Starting dashboard stats broadcast loop (interval: %ds)", interval_seconds)

    # Import here to avoid circular dependency
    from app.core.connection_manager import manager
    from app.core.database import SessionLocal

    while _broadcast_running:
        try:
            # Create a new database session for this iteration
            async with SessionLocal() as db:
                service = DashboardStatsService(db)
                stats = await service.get_all_stats()

                # Broadcast via WebSocket
                await manager.emit_dashboard_stats(stats.model_dump(mode="json"))

        except Exception as e:
            logger.error("Error in dashboard stats broadcast: %s", e, exc_info=True)

        # Wait for next interval
        await asyncio.sleep(interval_seconds)

    logger.info("Dashboard stats broadcast loop stopped")

get_dashboard_stats(db) async

Get current dashboard statistics (REST endpoint).

This provides a synchronous way to fetch stats without WebSocket.

Parameters:

  • db (Annotated[AsyncSession, Depends(get_db)]) –

    Database session.

Returns:

  • DashboardStats ( DashboardStats ) –

    The current dashboard statistics.

Source code in app/api/v1/endpoints/dashboard.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@router.get("/stats", response_model=DashboardStats)
async def get_dashboard_stats(db: Annotated[AsyncSession, Depends(get_db)]) -> DashboardStats:
    """
    Get current dashboard statistics (REST endpoint).

    This provides a synchronous way to fetch stats without WebSocket.

    Args:
        db: Database session.

    Returns:
        DashboardStats: The current dashboard statistics.
    """
    service = DashboardStatsService(db)
    stats = await service.get_all_stats()
    return stats

start_broadcast_task(interval_seconds=5) async

Start the periodic broadcast background task.

Parameters:

  • interval_seconds (int, default: 5 ) –

    Broadcast interval in seconds.

Returns:

  • None –

    None

Source code in app/api/v1/endpoints/dashboard.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def start_broadcast_task(interval_seconds: int = 5) -> None:
    """
    Start the periodic broadcast background task.

    Args:
        interval_seconds: Broadcast interval in seconds.

    Returns:
        None
    """
    global _broadcast_task, _broadcast_running

    if _broadcast_task is not None and not _broadcast_task.done():
        logger.warning("Dashboard broadcast task already running")
        return

    _broadcast_running = True
    _broadcast_task = asyncio.create_task(broadcast_dashboard_stats_loop(interval_seconds))
    logger.info("Dashboard broadcast task started")

stop_broadcast_task() async

Stop the periodic broadcast background task.

Returns:

  • None –

    None

Source code in app/api/v1/endpoints/dashboard.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def stop_broadcast_task() -> None:
    """
    Stop the periodic broadcast background task.

    Returns:
        None
    """
    global _broadcast_task, _broadcast_running

    if _broadcast_task is None:
        return

    _broadcast_running = False

    # Wait for task to finish (should exit quickly after flag set)
    try:
        if _broadcast_task:
            await asyncio.wait_for(_broadcast_task, timeout=10.0)
    except (asyncio.TimeoutError, asyncio.CancelledError):
        if _broadcast_task:
            logger.warning("Dashboard broadcast task did not stop gracefully, cancelling")
            _broadcast_task.cancel()

    _broadcast_task = None
    logger.info("Dashboard broadcast task stopped")