Function get_merge_queue_stats

ci3/ci-metrics/github_data.py:637–666

Get merge queue failure rate by day. Triggers backfill if needed.

(date_from: str, date_to: str)

def get_merge_queue_stats(date_from: str, date_to: str) -> dict:
    """Get merge queue failure rate by day. Triggers backfill if needed."""
    # Ensure data is populated (async after first load)
    import db
    conn = db.get_db()
    count = conn.execute('SELECT COUNT(*) as c FROM merge_queue_daily').fetchone()['c']
    if count == 0:
        ensure_merge_queue_data()  # block on first load
    else:
        threading.Thread(target=ensure_merge_queue_data, daemon=True).start()

    rows = db.query(
        'SELECT date, total, success, failure, cancelled, in_progress '
        'FROM merge_queue_daily WHERE date >= ? AND date <= ? ORDER BY date',
        (date_from, date_to))

    total_runs = sum(r['total'] for r in rows)
    total_fail = sum(r['failure'] for r in rows)
    total_success = sum(r['success'] for r in rows)

    return {
        'by_date': rows,
        'summary': {
            'total_runs': total_runs,
            'total_success': total_success,
            'total_failure': total_fail,
            'failure_rate': round(total_fail / max(total_runs, 1) * 100, 1),
            'days': len([r for r in rows if r['total'] > 0]),
        },
    }