Replace stats-functions, lint all functions

.github/dependabot.yml

@@ -46,20 +46,6 @@ updates:
       - dependency-name: "*"
         update-types: ["version-update:semver-major"]
 
-  - package-ecosystem: "pip"
-    directory: "/stats-functions"
-    schedule:
-      interval: "weekly"
-    groups:
-      patch-updates:
-        patterns:
-          - "*"
-        update-types:
-          - "patch"
-    ignore:
-      - dependency-name: "*"
-        update-types: ["version-update:semver-major"]
-
   - package-ecosystem: "github-actions"
     # check for updates to github actions monthly
     directory: "/"

.github/workflows/lint-test-function.yml

@@ -38,11 +38,11 @@ jobs:
 
           python -m pip install --upgrade pip
           pip install -r "$SOURCE_DIR/requirements.txt"
-          pip install -r "$SOURCE_DIR/requirements-dev.txt"
 
-      # - name: Lint
-      #   run: |
-      #     ruff check --output-format=github stats-functions/${{ inputs.function_name }}/src
+      - name: Lint
+        working-directory: stats-functions/${{ inputs.function_name }}
+        run: |
+          ruff check --output-format=github src
 
       - name: Test with pytest
         working-directory: stats-functions/${{ inputs.function_name }}

.gitignore

@@ -4,6 +4,9 @@
 # pytest cache
 **/.pytest_cache/
 
+# coverage data
+**/.coverage
+
 # ruff cache
 **/.ruff_cache/
 

stats-functions/README.md

@@ -1,7 +1,3 @@
-# Stats Functions package
-
-The stats-function package contains shared configuration and utilities that can be used with any cloud functions.
-
 # Stats Functions
 
 Other directories contain python source code for production cloud functions which collect arXiv site usage data and persist it to the `stats-db.site_usage` database. See below for a description of each. All are cron jobs implemented as GCP Cloud Functions with pubsub triggers. Trigger messages are published by GCP Scheduler Jobs. All infrastructure as code can be found in `terraform/`.

stats-functions/aggregate_hourly_downloads/Makefile

@@ -0,0 +1,15 @@
+VENV := .venv
+PIP := $(VENV)/bin/pip
+
+.PHONY: venv lint test
+
+venv:
+	python3 -m venv $(VENV)
+	$(PIP) install --upgrade pip
+	$(PIP) install -r src/requirements.txt
+
+lint:
+	$(VENV)/bin/ruff check src
+
+test:
+	$(VENV)/bin/pytest tests/ --cov=src/ --cov-report=html --cov-fail-under=80

stats-functions/aggregate_hourly_downloads/src/config.py

@@ -1,10 +1,10 @@
-from typing import Optional
-from arxiv_functions.config import FunctionConfig, DatabaseConfig
+
+from arxiv_functions.config import DatabaseConfig, FunctionConfig
 
 
 class Config(FunctionConfig):
-    read_db: Optional[DatabaseConfig] = None
-    write_db: Optional[DatabaseConfig] = None
+    read_db: DatabaseConfig | None = None
+    write_db: DatabaseConfig | None = None
 
     max_event_age_in_minutes: int = 50
     batch_size_for_category_query: int = 10000

stats-functions/aggregate_hourly_downloads/src/entities.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, String, Integer
+from sqlalchemy import Column, Integer, String
 from sqlalchemy.orm import declarative_base
 
 ReadBase = declarative_base()

stats-functions/aggregate_hourly_downloads/src/main.py

@@ -1,40 +1,34 @@
-import os
 import logging
-from typing import Set, Dict, List, Tuple, Any, Union
+import os
 from datetime import datetime, timedelta, timezone
+from typing import Any
 
 import functions_framework
+from arxiv.identifier import Identifier, IdentifierException
+from arxiv_functions.exception import NoRetryError
+from arxiv_functions.utils import (
+    event_time_exceeds_retry_window,
+    get_engine_unix_socket,
+    parse_cloud_event_time,
+    set_up_cloud_logging,
+)
 from cloudevents.http import CloudEvent
-
 from google.cloud import bigquery
 from google.cloud.bigquery.table import RowIterator, _EmptyRowIterator
-
 from sqlalchemy import Row
-from sqlalchemy.orm import sessionmaker, aliased
+from sqlalchemy.orm import aliased, sessionmaker
+from stats_entities.site_usage import HourlyDownloads
 
 from config import get_config
 from entities import DocumentCategory, Metadata
 from models import (
-    PaperCategories,
-    DownloadData,
+    AggregationResult,
     DownloadCounts,
+    DownloadData,
     DownloadKey,
-    AggregationResult,
-)
-
-from stats_entities.site_usage import HourlyDownloads
-
-from arxiv_functions.exception import NoRetryError
-from arxiv_functions.utils import (
-    set_up_cloud_logging,
-    get_engine_unix_socket,
-    event_time_exceeds_retry_window,
-    parse_cloud_event_time,
+    PaperCategories,
 )
 
-from arxiv.identifier import Identifier, IdentifierException
-
-
 config = get_config(os.getenv("ENV"))
 
 logger = logging.getLogger(__name__)
@@ -48,9 +42,9 @@
 
 
 def process_table_rows(
-    rows: Union[RowIterator, _EmptyRowIterator],
-) -> Tuple[
-    Any, Set[str], Set[datetime], int, int
+    rows: RowIterator | _EmptyRowIterator,
+) -> tuple[
+    Any, set[str], set[datetime], int, int
 ]:  # Changed return types to accommodate generator
     """
     processes rows of data from bigquery
@@ -85,14 +79,14 @@ def download_data_generator():
             except IdentifierException:
                 counts["bad_id"] += 1
                 continue
-            except Exception:
+            except Exception:  # noqa: BLE001 - one bad row must not abort the whole batch
                 counts["problem"] += 1
                 continue
 
     return download_data_generator(), paper_ids, time_periods, counts
 
 
-def get_paper_categories(paper_ids: Set[str]) -> List[Row[Tuple[str, str, int]]]:
+def get_paper_categories(paper_ids: set[str]) -> list[Row[tuple[str, str, int]]]:
     meta = aliased(Metadata)
     dc = aliased(DocumentCategory)
 
@@ -120,10 +114,10 @@ def get_paper_categories(paper_ids: Set[str]) -> List[Row[Tuple[str, str, int]]]
 
 
 def process_paper_categories(
-    data: List[Row[Tuple[str, str, int]]],
-) -> Dict[str, PaperCategories]:
+    data: list[Row[tuple[str, str, int]]],
+) -> dict[str, PaperCategories]:
     # format paper categories into dictionary
-    paper_categories: Dict[str, PaperCategories] = {}
+    paper_categories: dict[str, PaperCategories] = {}
     for row in data:
         paper_id, cat, is_primary = row
         entry = paper_categories.setdefault(paper_id, PaperCategories(paper_id))
@@ -136,14 +130,14 @@ def process_paper_categories(
 
 
 def aggregate_data(
-    download_data: List[DownloadData],
-    paper_categories: Dict[str, PaperCategories],
-) -> Dict[DownloadKey, DownloadCounts]:
+    download_data: list[DownloadData],
+    paper_categories: dict[str, PaperCategories],
+) -> dict[DownloadKey, DownloadCounts]:
     """creates a dictionary of download counts by time, country, download type, and category
     goes through each download entry, matches it with its caegories and adds the number of downloads to the count
     """
     logger.info("Aggregating download data")
-    all_data: Dict[DownloadKey, DownloadCounts] = {}
+    all_data: dict[DownloadKey, DownloadCounts] = {}
     missing_data_count = 0
 
     for entry in download_data:
@@ -185,8 +179,8 @@ def aggregate_data(
 
 
 def insert_into_database(
-    aggregated_data: Dict[DownloadKey, DownloadCounts],
-    time_periods: Set[datetime],  # Changed to Set
+    aggregated_data: dict[DownloadKey, DownloadCounts],
+    time_periods: set[datetime],  # Changed to Set
 ) -> int:
     """adds the data from an hour of downloads into the database
     uses bulk insert and update statements to increase efficiency
@@ -222,7 +216,7 @@ def insert_into_database(
 
 
 def perform_aggregation(
-    rows: Union[RowIterator, _EmptyRowIterator],
+    rows: RowIterator | _EmptyRowIterator,
 ) -> AggregationResult:
     logger.info("Processing results of log query")
     data_gen, paper_ids, time_periods, counts = process_table_rows(rows)
@@ -383,7 +377,7 @@ def aggregate_hourly_downloads(cloud_event: CloudEvent):
         if read_engine:
             try:
                 read_engine.dispose()
-            except Exception as dispose_err:
+            except Exception as dispose_err:  # noqa: BLE001 - a dispose failure must not mask the original error
                 logger.warning(f"Failed to dispose read_engine: {dispose_err}")
             finally:
                 read_engine = None
@@ -392,11 +386,11 @@ def aggregate_hourly_downloads(cloud_event: CloudEvent):
         if write_engine:
             try:
                 write_engine.dispose()
-            except Exception as dispose_err:
+            except Exception as dispose_err:  # noqa: BLE001 - a dispose failure must not mask the original error
                 logger.warning(f"Failed to dispose write_engine: {dispose_err}")
             finally:
                 write_engine = None
                 WriteSessionFactory = None
 
         # reraise to log traceback
-        raise e
+        raise

stats-functions/aggregate_hourly_downloads/src/models.py

@@ -1,6 +1,6 @@
 import logging
-from typing import Set, Literal
 from datetime import datetime
+from typing import Literal
 
 from arxiv.taxonomy.category import Category
 from arxiv.taxonomy.definitions import CATEGORIES
@@ -14,7 +14,7 @@
 class PaperCategories:
     paper_id: str
     primary: Category
-    crosses: Set[Category]
+    crosses: set[Category]
 
     def __init__(self, id: str):
         self.paper_id = id

stats-functions/aggregate_hourly_downloads/src/requirements.in

@@ -6,4 +6,7 @@ sqlalchemy>=2.0.26
 pydantic==2.*
 stats-entities @ git+https://github.com/arXiv/stats.git@main#subdirectory=stats-entities
 arxiv-functions @ git+https://github.com/arXiv/arxiv-base.git@develop#subdirectory=arxiv-functions
-pymysql>=1.1.0
+pymysql>=1.1.0
+pytest
+pytest-cov
+ruff