Build a Custom Developer Productivity Dashboard in Python: Track PR Cycle Time, Review Metrics, and Team Velocity with FastAPI

You’re running a dev team, and you have no idea how long PRs actually sit in review. Sound familiar?

Most teams rely on gut feel or manual Slack check-ins to measure productivity. That’s not engineering — that’s hoping. We ran into the same problem with our distributed team in Ho Chi Minh City and Can Tho. So we built a real-time developer productivity dashboard that surfaces hard metrics: PR cycle time, review latency, deployment frequency, and individual contributor velocity.

Building a Remote Engineering Team in Vietnam: A Step-by-Step Guide

Everything you need to know about building an effective remote engineering team in Vietnam. From hiring practices to… ...

Here’s the exact architecture and code we used. You can replicate this in an afternoon.

Why Build a Custom Dashboard Instead of Using Off-the-Shelf Tools?

Tools like Linear, Jira, and GitHub Insights give you some data. But they don’t let you define what “productivity” means for your specific team. You want to track how long a PR waits for a first review? How many re-requests happen per reviewer? Which developers consistently ship on Fridays vs Mondays?

Stop Reviewing Code Like It’s 2019: Why Your Team Needs AI Code Review Automation Tools Now

TL;DR: Manual code review is slow, inconsistent, and expensive. Modern AI code review automation tools catch bugs 40%… ...

Off-the-shelf tools won’t give you that level of granularity without heavy customization. A custom dashboard gives you full control over the metrics, the data sources, and the visualizations.

Plus, it’s a fun project. Let’s build it.

The Architecture: What We’re Building

We’ll build a three-tier system:

Data Collector — A Python service that polls the GitHub API for PR events, review comments, and merge timestamps
API Layer — A FastAPI backend that serves aggregated metrics
Frontend — A lightweight Chart.js dashboard that renders in the browser

We’ll use Redis for caching and real-time data freshness. PostgreSQL for persistent storage of historical metrics.

Here’s the high-level flow:


GitHub API → Data Collector (Python) → PostgreSQL + Redis → FastAPI → Chart.js Dashboard

Step 1: The Data Collector — Pulling PR Metrics from GitHub

This is the core of the system. We need to extract PR cycle time — the time between when a PR is opened and when it’s merged. We also need review latency — the time between when a review is requested and when the first review comment is posted.

python
import requests
from datetime import datetime, timezone
from typing import List, Dict, Optional
import time
import psycopg2
from psycopg2.extras import RealDictCursor
import redis
import json

class GitHubPRCollector:
    def __init__(self, token: str, org: str, repo: str):
        self.headers = {"Authorization": f"Bearer {token}"}
        self.base_url = f"https://api.github.com/repos/{org}/{repo}"
        self.org = org
        self.repo = repo
        self.redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
        self.db_conn = psycopg2.connect(
            host="localhost",
            dbname="dev_metrics",
            user="admin",
            password="your_password"
        )

    def fetch_merged_prs(self, since: str, page: int = 1) -> List[Dict]:
        """Fetch merged PRs since a given date."""
        url = f"{self.base_url}/pulls"
        params = {
            "state": "closed",
            "sort": "updated",
            "direction": "desc",
            "per_page": 100,
            "page": page,
            "since": since
        }
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        prs = resp.json()
        return [pr for pr in prs if pr.get("merged_at") is not None]

    def get_pr_reviews(self, pr_number: int) -> List[Dict]:
        """Fetch all reviews for a given PR."""
        url = f"{self.base_url}/pulls/{pr_number}/reviews"
        resp = requests.get(url, headers=self.headers)
        resp.raise_for_status()
        return resp.json()

    def compute_pr_metrics(self, pr: Dict) -> Dict:
        """Compute cycle time and review latency for a single PR."""
        created_at = datetime.fromisoformat(pr["created_at"].replace("Z", "+00:00"))
        merged_at = datetime.fromisoformat(pr["merged_at"].replace("Z", "+00:00"))
        cycle_time_hours = (merged_at - created_at).total_seconds() / 3600

        reviews = self.get_pr_reviews(pr["number"])
        first_review_time = None
        if reviews:
            first_review = min(reviews, key=lambda r: r["submitted_at"])
            review_time = datetime.fromisoformat(first_review["submitted_at"].replace("Z", "+00:00"))
            first_review_time = (review_time - created_at).total_seconds() / 3600

        return {
            "pr_number": pr["number"],
            "author": pr["user"]["login"],
            "title": pr["title"],
            "created_at": pr["created_at"],
            "merged_at": pr["merged_at"],
            "cycle_time_hours": round(cycle_time_hours, 2),
            "first_review_latency_hours": round(first_review_time, 2) if first_review_time else None,
            "review_count": len(reviews),
            "additions": pr.get("additions", 0),
            "deletions": pr.get("deletions", 0)
        }

    def store_metrics(self, metrics: Dict):
        """Store metrics in PostgreSQL and cache in Redis."""
        cursor = self.db_conn.cursor()
        cursor.execute("""
            INSERT INTO pr_metrics 
            (pr_number, author, title, created_at, merged_at, cycle_time_hours, 
             first_review_latency_hours, review_count, additions, deletions)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (pr_number) DO UPDATE SET
                cycle_time_hours = EXCLUDED.cycle_time_hours,
                first_review_latency_hours = EXCLUDED.first_review_latency_hours,
                review_count = EXCLUDED.review_count
        """, (
            metrics["pr_number"], metrics["author"], metrics["title"],
            metrics["created_at"], metrics["merged_at"],
            metrics["cycle_time_hours"], metrics["first_review_latency_hours"],
            metrics["review_count"], metrics["additions"], metrics["deletions"]
        ))
        self.db_conn.commit()
        cursor.close()

        # Cache the latest metrics in Redis for real-time dashboard access
        cache_key = f"pr:{metrics['pr_number']}:metrics"
        self.redis_client.setex(cache_key, 3600, json.dumps(metrics))

    def run_collection(self, since: str = "2025-01-01T00:00:00Z"):
        """Main collection loop."""
        page = 1
        while True:
            prs = self.fetch_merged_prs(since, page)
            if not prs:
                break
            for pr in prs:
                metrics = self.compute_pr_metrics(pr)
                self.store_metrics(metrics)
                print(f"Collected PR #{pr['number']} — cycle time: {metrics['cycle_time_hours']}h")
                time.sleep(0.5)  # Be nice to GitHub's rate limits
            page += 1
        print("Collection complete.")

This collector runs as a scheduled job — we use a cron expression that fires every 15 minutes. It only fetches PRs merged since the last run, so it’s incremental.

Pro tip: GitHub’s API rate limits at 5,000 requests per hour for authenticated users. Our collector makes about 2 requests per PR (one for the PR list, one for reviews), so we can handle about 2,500 PRs per hour. That’s plenty for most teams.