Merge pull request #13 from gavinelder/ge/feat/daily-stats

Author: ewels

app/db.py

@@ -7,6 +7,7 @@
 
 import pandas as pd
 
+from sqlalchemy import text
 from sqlmodel import create_engine, Field, select, Session, SQLModel
 
 sql_url = os.getenv("DATABASE_URL")
@@ -38,6 +39,27 @@ class VisitStats(SQLModel, table=True):
     count: int
 
 
+class VisitDailyStats(SQLModel, table=True):
+    """
+    Daily aggregated visit statistics.
+
+    Populated by a daily task that rolls up from multiqc_api_visits_stats.
+    """
+
+    __tablename__ = "multiqc_api_visits_daily_stats"
+
+    day: datetime.date = Field(primary_key=True)
+    version_multiqc: str = Field(primary_key=True)
+    version_python: str = Field(primary_key=True)
+    operating_system: str = Field(primary_key=True)
+    is_docker: bool = Field(primary_key=True)
+    is_singularity: bool = Field(primary_key=True)
+    is_conda: bool = Field(primary_key=True)
+    is_ci: bool = Field(primary_key=True)
+    is_uv: bool = Field(primary_key=True)
+    count: int
+
+
 class DownloadStats(SQLModel, table=True):
     """
     Daily download statistics.
@@ -97,12 +119,12 @@ def get_download_stats(
     with Session(engine) as session:
         statement = select(DownloadStats)
         if start:
-            statement.where(DownloadStats.date >= start)  # type: ignore
+            statement = statement.where(DownloadStats.date >= start)  # type: ignore
         if end:
-            statement.where(DownloadStats.date <= end)  # type: ignore
+            statement = statement.where(DownloadStats.date <= end)  # type: ignore
         if limit:
-            statement.limit(limit)
-        statement.order_by(DownloadStats.date.desc())  # type: ignore
+            statement = statement.limit(limit)
+        statement = statement.order_by(DownloadStats.date.desc())  # type: ignore
         return session.exec(statement).all()
 
 
@@ -114,6 +136,76 @@ def insert_visit_stats(visit_stats: pd.DataFrame):
         session.commit()
 
 
+def has_daily_stats_for_date(day: datetime.date) -> bool:
+    """Check if daily stats already exist for a given date."""
+    with Session(engine) as session:
+        result = session.exec(
+            select(VisitDailyStats).where(VisitDailyStats.day == day).limit(1)
+        ).first()
+        return result is not None
+
+
+def aggregate_visits_for_date(target_date: datetime.date) -> int:
+    """
+    Aggregate visit stats for a specific date and upsert into daily stats table.
+
+    Returns the number of rows inserted/updated.
+    """
+    query = text("""
+        INSERT INTO multiqc_api_visits_daily_stats (
+            day,
+            version_multiqc,
+            version_python,
+            operating_system,
+            is_docker,
+            is_singularity,
+            is_conda,
+            is_ci,
+            is_uv,
+            count
+        )
+        SELECT
+            :target_date AS day,
+            version_multiqc,
+            version_python,
+            operating_system,
+            is_docker,
+            is_singularity,
+            is_conda,
+            is_ci,
+            is_uv,
+            SUM(count) AS count
+        FROM multiqc_api_visits_stats
+        WHERE start >= :target_date AND start < :target_date + INTERVAL '1 day'
+        GROUP BY
+            version_multiqc,
+            version_python,
+            operating_system,
+            is_docker,
+            is_singularity,
+            is_conda,
+            is_ci,
+            is_uv
+        ON CONFLICT (
+            day,
+            version_multiqc,
+            version_python,
+            operating_system,
+            is_docker,
+            is_singularity,
+            is_conda,
+            is_ci,
+            is_uv
+        )
+        DO UPDATE SET count = EXCLUDED.count
+    """)
+
+    with engine.connect() as conn:
+        result = conn.execute(query, {"target_date": target_date})
+        conn.commit()
+        return result.rowcount
+
+
 def insert_download_stats(df: pd.DataFrame) -> pd.DataFrame:
     # df has "date" as an index. Re-adding it as a separate field with a type datetime
     df["date"] = pd.to_datetime(df.index)

app/main.py

@@ -24,6 +24,7 @@
 
 from app import __version__, db, models
 from app.downloads import daily
+from app.db import aggregate_visits_for_date, has_daily_stats_for_date
 
 
 logger = logging.getLogger("multiqc_api")
@@ -346,6 +347,36 @@ async def update_downloads():
     _update_download_stats()
 
 
+def _aggregate_daily_visits():
+    """
+    Aggregate yesterday's visit stats into the daily stats table.
+    Skips if yesterday's data has already been aggregated.
+    """
+    yesterday = datetime.date.today() - datetime.timedelta(days=1)
+    if has_daily_stats_for_date(yesterday):
+        logger.info(f"Daily stats for {yesterday} already exist, skipping")
+        return
+    logger.info(f"Aggregating visit stats for {yesterday}...")
+    try:
+        rows = aggregate_visits_for_date(yesterday)
+        logger.info(f"Successfully aggregated {rows} rows for {yesterday}")
+    except Exception as e:
+        logger.error(f"Failed to aggregate visit stats: {e}")
+
+
+@app.on_event("startup")
+@repeat_every(
+    seconds=60 * 60 * 24,  # every day
+    wait_first=True,
+    logger=logger,
+)
+async def aggregate_daily_visits():
+    """
+    Repeated task to aggregate yesterday's visits into daily stats.
+    """
+    _aggregate_daily_visits()
+
+
 @app.post("/persist_visits")
 async def persist_visits_endpoint():
     try: