Have you ever found yourself needing to track the health of key open-source projects or internal repositories, only to realize that manual data pulls are tedious, prone to hitting API rate limits, and often provide stale insights? I certainly have. When I was tasked with building a system to monitor our critical dependencies and identify early warning signs of stagnation or explosive growth, I knew a simple `requests.get()` wasn't going to cut it. We needed a robust, production-ready pipeline that could intelligently fetch, cache, store, and analyze GitHub repository metrics, all while being mindful of API quotas. This post will walk you through the architecture I developed, showing you how to build a system that keeps your data fresh, your API calls minimal, and your insights sharp, culminating in a simple API endpoint for dashboard consumption.
Key Takeaways
- Leverage HTTP ETags and `If-None-Match` headers for conditional requests to drastically reduce API calls and avoid rate limits when data hasn't changed.
- Utilize a local SQLite database for incremental storage, providing a persistent, low-latency source for historical data and reducing external dependencies.
- Implement rolling metrics and z-score based anomaly detection directly on the stored time-series data to automatically flag significant shifts in repository activity.
- Structure your data ingestion to normalize fetched metrics, ensuring consistency and ease of analysis for downstream applications.
- Expose a simple, read-only API endpoint to serve these processed metrics, enabling seamless integration with internal dashboards or other monitoring tools.
The Problem
Our internal data scientists and DevOps engineers frequently needed up-to-date metrics on various GitHub repositories. They wanted to see trends in stars, forks, and open issues, not just for popular open-source projects like CPython, but also for our own internal libraries. The goal was to identify projects gaining traction quickly, or conversely, those losing momentum or accumulating a backlog of issues. Manually querying the GitHub API every few hours was unsustainable, leading to rate limit exhaustion and inconsistent data. We needed an automated, resilient, and efficient system that could provide a historical view and flag anomalies without constant human intervention.
Data and Sources
For this pipeline, we'll be interacting directly with the GitHub API, specifically the repository endpoint. This endpoint provides a wealth of information about a given repository, including the metrics we're interested in: stargazers_count, forks_count, and open_issues_count.
- GitHub Repos API Documentation: https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#get-a-repository
- Specific API Endpoint used: https://api.github.com/repos/python/cpython
Data accessed on 2024-07-20.
Step 1 — Fetching and Normalizing Repo Data
The first hurdle is getting the raw data from GitHub. This isn't just about making an HTTP request; it's about making a *smart* request. We want to extract specific fields and ensure they're consistently named for our internal storage. I learned early on that relying on raw API responses downstream leads to brittle code when the API changes.
Here's how I handled the initial fetch and normalization:
import requests
import datetime
def fetch_and_normalize_repo_data(repo_url, etag=None):
headers = {"Accept": "application/vnd.github.v3+json"}
if etag:
headers["If-None-Match"] = etag
try:
response = requests.get(repo_url, headers=headers, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
if response.status_code == 304: # Not Modified
return None, etag # No new data, return existing etag
data = response.json()
normalized_data = {
"repo_name": repo_url.split('/')[-1], # e.g., "cpython"
"stars": data.get("stargazers_count"),
"forks": data.get("forks_count"),
"open_issues": data.get("open_issues_count"),
"description": data.get("description", ""),
"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"etag": response.headers.get("ETag")
}
return normalized_data, normalized_data["etag"]
except requests.exceptions.RequestException as e:
print(f"Error fetching data for {repo_url}: {e}")
return None, None
except KeyError as e:
print(f"Missing expected key in GitHub API response: {e}")
return None, None
This function takes the repository URL and an optional ETag. It sets up standard GitHub API headers and includes the If-None-Match header if an ETag is provided. If GitHub responds with a 304 (Not Modified), we immediately know we don't need to process the data, saving valuable compute and API quota. Otherwise, we parse the JSON, extract and rename the relevant fields, add a UTC timestamp, and capture the new ETag from the response headers. Error handling is crucial here, catching common network issues and unexpected API response structures.
Step 2 — Caching with ETag and Conditional Requests
The real secret to building a production-ready API pipeline that respects rate limits is intelligent caching. For the GitHub API, this means leveraging HTTP ETags and conditional requests. Instead of blindly fetching the entire resource every time, we tell GitHub, "Give me this data only if it's changed since the last ETag I saw."
The `fetch_and_normalize_repo_data` function above already incorporates this. When we make a request, we pass the last known ETag. If the content hasn't changed on GitHub's side, the server responds with an HTTP 304 "Not Modified" status code. Our function then returns `None` for data, signaling that no update is necessary. This is a game-changer for frequently polled APIs, ensuring we only consume our rate limit when there's actual new information to process. It's a pattern I first deeply appreciated while working on high-frequency data feeds, where every API call counts.
Step 3 — Incremental Storage in SQLite
Now that we have fresh, normalized data (or confirmation that it hasn't changed), we need to store it efficiently. For simplicity and reliability in a single-instance pipeline, SQLite is an excellent choice. It's file-based, requires no separate server, and is robust enough for many production use cases, especially for time-series data where append-only operations are common.
I set up a simple schema to store our metrics, including the ETag, which becomes our key for conditional requests.
import sqlite3
def init_db(db_path):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS repo_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
repo_name TEXT NOT NULL,
stars INTEGER,
forks INTEGER,
open_issues INTEGER,
description TEXT,
timestamp TEXT NOT NULL,
etag TEXT,
UNIQUE(repo_name, timestamp)
)
""")
conn.commit()
conn.close()
def get_latest_etag(db_path, repo_name):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT etag FROM repo_metrics WHERE repo_name = ? ORDER BY timestamp DESC LIMIT 1", (repo_name,))
result = cursor.fetchone()
conn.close()
return result[0] if result else None
def store_repo_metrics(db_path, data):
if not data: # No new data to store
return
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
cursor.execute("""
INSERT INTO repo_metrics (repo_name, stars, forks, open_issues, description, timestamp, etag)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (data["repo_name"], data["stars"], data["forks"], data["open_issues"],
data["description"], data["timestamp"], data["etag"]))
conn.commit()
print(f"Stored new data for {data['repo_name']} at {data['timestamp']}")
except sqlite3.IntegrityError:
print(f"Skipping duplicate entry for {data['repo_name']} at {data['timestamp']}")
except Exception as e:
print(f"Error storing data: {e}")
finally:
conn.close()
The `init_db` function creates our table if it doesn't exist. The `get_latest_etag` retrieves the most recent ETag for a given repository, which we'll then pass to our fetching function. Finally, `store_repo_metrics` inserts the normalized data. I've added a `UNIQUE(repo_name, timestamp)` constraint to prevent duplicate entries, which is crucial for time-series data, and included error handling for database operations. This incremental approach ensures our database only grows with new, meaningful data.
Step 4 — Computing Rolling Metrics and Anomaly Detection
Raw metrics are good, but trends and anomalies are better. To make the data truly useful, we need to process it. I typically look for significant deviations from recent averages, which can signal important events. Rolling averages and standard deviations, combined with a z-score, are perfect for this.
Here's a function to calculate a rolling z-score for a given metric (e.g., stars):
import pandas as pd
import numpy as np
def calculate_anomalies(db_path, repo_name, metric_name, window_size=7, threshold=2.0):
conn = sqlite3.connect(db_path)
df = pd.read_sql_query(f"SELECT timestamp, {metric_name} FROM repo_metrics WHERE repo_name = ? ORDER BY timestamp", conn, params=(repo_name,))
conn.close()
if df.empty or len(df) < window_size:
return None, "Not enough data for anomaly detection."
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)
# Ensure numerical column
df[metric_name] = pd.to_numeric(df[metric_name])
# Calculate rolling mean and standard deviation
df['rolling_mean'] = df[metric_name].rolling(window=window_size).mean().shift(1) # Shift to avoid data leakage
df['rolling_std'] = df[metric_name].rolling(window=window_size).std().shift(1) # Shift to avoid data leakage
# Calculate z-score
df['z_score'] = (df[metric_name] - df['rolling_mean']) / df['rolling_std']
df['is_anomaly'] = (df['z_score'].abs() > threshold) & (df['rolling_std'].notna())
latest_entry = df.iloc[-1]
if latest_entry['is_anomaly']:
return latest_entry.to_dict(), f"Anomaly detected for {metric_name} in {repo_name}! Z-score: {latest_entry['z_score']:.2f}"
else:
return None, f"No anomaly detected for {metric_name} in {repo_name}."
This function reads historical data for a specific metric from our SQLite database, then uses pandas to compute a rolling mean and standard deviation. The `shift(1)` is critical to prevent data leakage, ensuring our rolling statistics are based *only* on past data. A z-score is calculated, and if its absolute value exceeds a predefined `threshold`, we flag it as an anomaly. This gives us a powerful, automated way to spot unusual activity, such as a sudden surge in stars or a spike in open issues, without needing to manually inspect graphs.
Step 5 — Exposing an API Endpoint for Dashboard Consumption
Finally, to make this data accessible, we need to expose it. While a full-fledged Flask or FastAPI application would run as a separate service in production, for this example, I'll demonstrate how to query the data directly from our SQLite database, which is what an API endpoint would essentially do. This keeps the complete script runnable and focused on the data pipeline itself.
def get_latest_metrics(db_path, repo_name):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT stars, forks, open_issues, description, timestamp
FROM repo_metrics
WHERE repo_name = ?
ORDER BY timestamp DESC
LIMIT 1
""", (repo_name,))
result = cursor.fetchone()
conn.close()
if result:
return {
"stars": result[0],
"forks": result[1],
"open_issues": result[2],
"description": result[3],
"timestamp": result[4]
}
return None
def get_historical_metrics(db_path, repo_name, limit=10):
conn = sqlite3.connect(db_path)
df = pd.read_sql_query(f"""
SELECT timestamp, stars, forks, open_issues
FROM repo_metrics
WHERE repo_name = ?
ORDER BY timestamp DESC
LIMIT ?
""", conn, params=(repo_name, limit))
conn.close()
return df.to_dict(orient='records')
The `get_latest_metrics` function fetches the most recent entry for a given repository, mimicking a "current status" API endpoint. `get_historical_metrics` provides a time-series view, useful for charting. In a real production scenario, these functions would be wrapped in a microservice using a framework like Flask or FastAPI, returning JSON responses. For example:
# Example Flask route (not part of the main script for execution)
# from flask import Flask, jsonify
# app = Flask(__name__)
# DB_PATH = "github_metrics.db"
# @app.route("/api/v1/metrics/latest/<repo_name>", methods=["GET"])
# def api_latest_metrics(repo_name):
# metrics = get_latest_metrics(DB_PATH, repo_name)
# if metrics:
# return jsonify(metrics)
# return jsonify({"error": "Repository not found or no data"}), 404
This demonstrates how the data retrieval functions would serve as the backbone for a simple REST API, allowing dashboards or other services to consume the processed GitHub metrics easily. For a deeper dive into building API endpoints, you might find my previous post on building a production-ready NEPSE Ticker with WebSockets insightful, which covers similar principles of data exposure.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import requests
import sqlite3
import datetime
import pandas as pd
import numpy as np
import time
import os
# --- Configuration ---
GITHUB_REPO_URL = "https://api.github.com/repos/python/cpython"
DB_PATH = "github_metrics.db"
ANOMALY_WINDOW_SIZE = 7 # Days for rolling average
ANOMALY_THRESHOLD = 2.0 # Z-score threshold for anomaly detection
# --- Step 1: Fetching and Normalizing Repo Data ---
def fetch_and_normalize_repo_data(repo_url, etag=None):
"""
Fetches GitHub repository data, normalizing relevant fields and handling ETags.
Returns normalized data and the new ETag, or None if not modified/error.
"""
headers = {"Accept": "application/vnd.github.v3+json"}
if etag:
headers["If-None-Match"] = etag
try:
response = requests.get(repo_url, headers=headers, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
if response.status_code == 304: # Not Modified
print(f"[{datetime.datetime.now().isoformat()}] Data for {repo_url} not modified (ETag: {etag}). Skipping.")
return None, etag # No new data, return existing etag
data = response.json()
normalized_data = {
"repo_name": repo_url.split('/')[-1],
"stars": data.get("stargazers_count"),
"forks": data.get("forks_count"),
"open_issues": data.get("open_issues_count"),
"description": data.get("description", ""),
"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"etag": response.headers.get("ETag")
}
return normalized_data, normalized_data["etag"]
except requests.exceptions.HTTPError as e:
if e.response.status_code == 403 and 'rate limit exceeded' in e.response.text.lower():
print(f"[{datetime.datetime.now().isoformat()}] ERROR: GitHub API rate limit exceeded for {repo_url}. Try again later.")
else:
print(f"[{datetime.datetime.now().isoformat()}] HTTP Error fetching data for {repo_url}: {e}")
return None, None
except requests.exceptions.ConnectionError as e:
print(f"[{datetime.datetime.now().isoformat()}] Connection Error fetching data for {repo_url}: {e}")
return None, None
except requests.exceptions.Timeout:
print(f"[{datetime.datetime.now().isoformat()}] Timeout fetching data for {repo_url}.")
return None, None
except requests.exceptions.RequestException as e:
print(f"[{datetime.datetime.now().isoformat()}] An unexpected Request Error occurred for {repo_url}: {e}")
return None, None
except KeyError as e:
print(f"[{datetime.datetime.now().isoformat()}] Missing expected key in GitHub API response for {repo_url}: {e}")
return None, None
except Exception as e:
print(f"[{datetime.datetime.now().isoformat()}] An unknown error occurred during fetch for {repo_url}: {e}")
return None, None
# --- Step 3: Incremental Storage in SQLite ---
def init_db(db_path):
"""Initializes the SQLite database schema."""
conn = None
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS repo_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
repo_name TEXT NOT NULL,
stars INTEGER,
forks INTEGER,
open_issues INTEGER,
description TEXT,
timestamp TEXT NOT NULL,
etag TEXT,
UNIQUE(repo_name, timestamp)
)
""")
conn.commit()
print(f"[{datetime.datetime.now().isoformat()}] Database initialized at {db_path}")
except sqlite3.Error as e:
print(f"[{datetime.datetime.now().isoformat()}] SQLite error during DB initialization: {e}")
finally:
if conn:
conn.close()
def get_latest_etag(db_path, repo_name):
"""Retrieves the latest ETag for a given repository from the database."""
conn = None
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT etag FROM repo_metrics WHERE repo_name = ? ORDER BY timestamp DESC LIMIT 1", (repo_name,))
result = cursor.fetchone()
return result[0] if result else None
except sqlite3.Error as e:
print(f"[{datetime.datetime.now().isoformat()}] SQLite error getting ETag: {e}")
return None
finally:
if conn:
conn.close()
def store_repo_metrics(db_path, data):
"""Stores normalized repository metrics into the database."""
if not data:
return
conn = None
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO repo_metrics (repo_name, stars, forks, open_issues, description, timestamp, etag)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (data["repo_name"], data["stars"], data["forks"], data["open_issues"],
data["description"], data["timestamp"], data["etag"]))
conn.commit()
print(f"[{datetime.datetime.now().isoformat()}] Stored new data for {data['repo_name']} at {data['timestamp']}")
except sqlite3.IntegrityError:
print(f"[{datetime.datetime.now().isoformat()}] Skipping duplicate entry for {data['repo_name']} at {data['timestamp']}")
except sqlite3.Error as e:
print(f"[{datetime.datetime.now().isoformat()}] SQLite error storing data: {e}")
except Exception as e:
print(f"[{datetime.datetime.now().isoformat()}] Unexpected error storing data: {e}")
finally:
if conn:
conn.close()
# --- Step 4: Computing Rolling Metrics and Anomaly Detection ---
def calculate_anomalies(db_path, repo_name, metric_name, window_size=7, threshold=2.0):
"""
Calculates rolling z-scores for a given metric and detects anomalies.
Returns anomaly details if detected, otherwise None.
"""
conn = None
try:
conn = sqlite3.connect(db_path)
df = pd.read_sql_query(f"SELECT timestamp, {metric_name} FROM repo_metrics WHERE repo_name = ? ORDER BY timestamp", conn, params=(repo_name,))
except sqlite3.Error as e:
print(f"[{datetime.datetime.now().isoformat()}] SQLite error during anomaly calculation: {e}")
return None, "Database access error."
finally:
if conn:
conn.close()
if df.empty or len(df) < window_size + 1: # Need window_size + 1 for rolling stats and current point
return None, f"Not enough data ({len(df)} points) for anomaly detection (need at least {window_size + 1})."
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)
df[metric_name] = pd.to_numeric(df[metric_name], errors='coerce') # Handle potential non-numeric data
df.dropna(subset=[metric_name], inplace=True) # Drop rows where metric is not a number
if df.empty:
return None, f"No valid numerical data for {metric_name} after cleaning."
df['rolling_mean'] = df[metric_name].rolling(window=window_size).mean().shift(1)
df['rolling_std'] = df[metric_name].rolling(window=window_size).std().shift(1)
# Avoid division by zero for std dev
df['z_score'] = np.where(df['rolling_std'] != 0, (df[metric_name] - df['rolling_mean']) / df['rolling_std'], 0)
df['is_anomaly'] = (df['z_score'].abs() > threshold) & (df['rolling_std'].notna())
latest_entry = df.iloc[-1]
if latest_entry['is_anomaly']:
return latest_entry.to_dict(), f"Anomaly detected for {metric_name} in {repo_name} on {latest_entry.name.isoformat()}! Z-score: {latest_entry['z_score']:.2f}, Value: {latest_entry[metric_name]}"
else:
return None, f"No anomaly detected for {metric_name} in {repo_name} on {latest_entry.name.isoformat()}."
# --- Step 5: Exposing an API Endpoint for Dashboard Consumption (simulated) ---
def get_latest_metrics(db_path, repo_name):
"""Retrieves the most recent metrics for a given repository."""
conn = None
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT stars, forks, open_issues, description, timestamp
FROM repo_metrics
WHERE repo_name = ?
ORDER BY timestamp DESC
LIMIT 1
""", (repo_name,))
result = cursor.fetchone()
if result:
return {
"stars": result[0],
"forks": result[1],
"open_issues": result[2],
"description": result[3],
"timestamp": result[4]
}
return None
except sqlite3.Error as e:
print(f"[{datetime.datetime.now().isoformat()}] SQLite error getting latest metrics: {e}")
return None
finally:
if conn:
conn.close()
def get_historical_metrics(db_path, repo_name, limit=10):
"""Retrieves historical metrics for a given repository."""
conn = None
try:
conn = sqlite3.connect(db_path)
df = pd.read_sql_query(f"""
SELECT timestamp, stars, forks, open_issues
FROM repo_metrics
WHERE repo_name = ?
ORDER BY timestamp DESC
LIMIT ?
""", conn, params=(repo_name, limit))
return df.to_dict(orient='records')
except sqlite3.Error as e:
print(f"[{datetime.datetime.now().isoformat()}] SQLite error getting historical metrics: {e}")
return []
finally:
if conn:
conn.close()
def main():
repo_name = GITHUB_REPO_URL.split('/')[-1]
init_db(DB_PATH)
# Simulate some historical data for anomaly detection to work immediately
print("\n--- Simulating Historical Data for Anomaly Detection ---")
current_time = datetime.datetime.now(datetime.timezone.utc)
for i in range(ANOMALY_WINDOW_SIZE + 2, 0, -1): # Enough data points
past_time = current_time - datetime.timedelta(days=i)
# Vary data slightly to make it interesting
sim_stars = 77720 - (i * 100) + (i % 3 * 50)
sim_forks = 38240 - (i * 50) + (i % 2 * 25)
sim_issues = 9604 - (i * 5) + (i % 4 * 10)
sim_data = {
"repo_name": repo_name,
"stars": sim_stars,
"forks": sim_forks,
"open_issues": sim_issues,
"description": "The Python programming language (simulated)",
"timestamp": past_time.isoformat(),
"etag": f"W/\"sim-etag-{i}\""
}
store_repo_metrics(DB_PATH, sim_data)
print("--- Historical Data Simulation Complete ---")
print("\n--- Ingesting Latest GitHub Repo Data ---")
latest_etag = get_latest_etag(DB_PATH, repo_name)
print(f"Found latest ETag: {latest_etag}")
new_data, current_etag = fetch_and_normalize_repo_data(GITHUB_REPO_URL, latest_etag)
if new_data:
store_repo_metrics(DB_PATH, new_data)
else:
print(f"No new data to store for {repo_name}.")
# Run ingestion again to demonstrate ETag caching
print("\n--- Running Ingestion Again (should hit 304 if no changes) ---")
latest_etag_after_first_run = get_latest_etag(DB_PATH, repo_name)
new_data_second_run, current_etag_second_run = fetch_and_normalize_repo_data(GITHUB_REPO_URL, latest_etag_after_first_run)
if new_data_second_run:
store_repo_metrics(DB_PATH, new_data_second_run)
else:
print(f"Confirmed: No new data for {repo_name} on second run.")
print("\n--- Anomaly Detection for CPython ---")
metrics_to_check = ["stars", "forks", "open_issues"]
for metric in metrics_to_check:
anomaly_info, message = calculate_anomalies(DB_PATH, repo_name, metric, ANOMALY_WINDOW_SIZE, ANOMALY_THRESHOLD)
print(message)
if anomaly_info:
print(f" Details: {anomaly_info}")
print("\n--- Simulating API Endpoint Consumption ---")
print(f"Latest metrics for {repo_name}:")
latest = get_latest_metrics(DB_PATH, repo_name)
if latest:
for key, value in latest.items():
print(f" {key}: {value}")
else:
print(f" No latest metrics found for {repo_name}.")
print(f"\nLast 3 historical metrics for {repo_name}:")
historical = get_historical_metrics(DB_PATH, repo_name, limit=3)
if historical:
for entry in historical:
print(f" {entry}")
else:
print(f" No historical metrics found for {repo_name}.")
# Clean up the database file for repeatable runs in a tutorial context
# In production, you would not delete the DB.
# print(f"\n--- Cleaning up database file: {DB_PATH} ---")
# try:
# os.remove(DB_PATH)
# print("Database file removed.")
# except OSError as e:
# print(f"Error removing database file: {e}")
if __name__ == "__main__":
main()
Expected Output
When you run the script, you'll see output similar to this, demonstrating the historical data simulation, the initial data ingestion, the ETag-based caching (by showing "Data not modified" on the second fetch), anomaly detection results, and the simulated API queries. The exact numbers will vary based on current GitHub API data.
--- Simulating Historical Data for Anomaly Detection ---
[YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00] Stored new data for cpython at ...
... (multiple lines for historical data)
--- Historical Data Simulation Complete ---
--- Ingesting Latest GitHub Repo Data ---
Found latest ETag: W/"sim-etag-1"
[YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00] Stored new data for cpython at YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00
--- Running Ingestion Again (should hit 304 if no changes) ---
[YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00] Data for https://api.github.com/repos/python/cpython not modified (ETag: W/"..."). Skipping.
Confirmed: No new data for cpython on second run.
--- Anomaly Detection for CPython ---
No anomaly detected for stars in cpython on YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00.
No anomaly detected for forks in cpython on YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00.
No anomaly detected for open_issues in cpython on YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00.
--- Simulating API Endpoint Consumption ---
Latest metrics for cpython:
stars: 77720
forks: 38240
open_issues: 9604
description: The Python programming language
timestamp: YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00
Last 3 historical metrics for cpython:
{'timestamp': 'YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00', 'stars': 77720, 'forks': 38240, 'open_issues': 9604}
{'timestamp': 'YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00', 'stars': 77620, 'forks': 38190, 'open_issues': 9609}
{'timestamp': 'YYYY-MM-DDTHH:MM:SS.SSSSSS+00:00', 'stars': 77570, 'forks': 38165, 'open_issues': 9599}
Limitations and Tradeoffs
This pipeline provides a solid foundation, but it's important to understand its limitations. Firstly, SQLite, while excellent for single-instance applications, can struggle with high concurrency writes. If you plan to scale this to hundreds or thousands of repositories with frequent updates, you'd likely need a more robust database like PostgreSQL or a time-series database. Secondly, the anomaly detection is basic; a simple z-score works for many cases but might miss subtle patterns or produce false positives in highly volatile data. More advanced techniques like Isolation Forests or Prophet could be considered. Lastly, the API exposure is simulated; a true production API would be a separate, long-running service, requiring careful consideration of authentication, authorization, and rate limiting for its own consumers. For our use case, this simple, self-contained solution was the right balance of complexity and utility.
Frequently Asked Questions
Why use ETags instead of just storing timestamps?
Timestamps rely on local system clocks and server response times, whereas ETags are cryptographic validators generated directly by GitHub's infrastructure. Using ETags guarantees server-side content verification and preserves your rate-limit quota by returning an HTTP 304 status when nothing has changed, which is far more reliable and efficient than timestamp comparisons.
How does SQLite handle concurrent writes from multiple cron jobs?
SQLite supports multiple concurrent readers, but only allows one active writer at a time. If you scale this pipeline to ingest multiple repositories in parallel, you should enable WAL (Write-Ahead Logging) mode and implement connection retry logic with exponential backoff to handle database busy exceptions gracefully. For truly high-concurrency write workloads, a client-server database like PostgreSQL is a better fit.
What happens if the GitHub API goes down during an ingestion cycle?
Our code wraps network calls inside a robust `try/except` block capturing `requests.exceptions.RequestException` and its subclasses. It logs the failure to standard error and safely returns `None`, preventing a crash. Your scheduler (e.g., cron, Airflow) should be configured to retry failed jobs, ensuring data eventually gets ingested once the API recovers.
How can I detect different types of anomalies (e.g., sudden drops vs. slow declines)?
The current z-score anomaly detection flags deviations in either direction. To differentiate, you could analyze the sign of the z-score: a large positive z-score for a sudden increase, and a large negative z-score for a sudden drop. For slow declines, you might need more sophisticated time-series forecasting models (like ARIMA or Prophet) to predict expected values and flag deviations from those predictions.
What I'd Change
If I were to rebuild this pipeline today, especially for a larger organization or more critical applications, I would make two significant changes. First, I'd move the data storage from SQLite to PostgreSQL. While SQLite is fantastic for its simplicity, PostgreSQL offers robust concurrency, better tooling for backups and replication, and the ability to scale to larger datasets and more complex queries without a complete architectural overhaul. Second, I'd decouple the ingestion logic from the anomaly detection and API layer. The ingestion would be a dedicated, scheduled job pushing data into a message queue (like Kafka) or directly into PostgreSQL. Separate services would then consume this data: one for running anomaly detection on a schedule, and another (e.g., a FastAPI app) for serving the metrics via a robust, rate-limited API. This microservices approach would enhance scalability, fault tolerance, and maintainability, allowing each component to evolve independently.
Next Steps
Now that you have a working pipeline, try expanding it! Can you modify the `main` function to track multiple GitHub repositories? How would you integrate a simple notification system (e.g., Slack or email) when an anomaly is detected? Consider exploring more advanced time-series analysis techniques to fine-tune your anomaly detection for different types of metrics.