Fix Ruff B017: replace blind pytest.raises(Exception) with specific Pandera exception

src/laptop_price/alerts.py

@@ -47,4 +47,4 @@ def main() -> None:
 
 
 if __name__ == "__main__":
-    main()
+    main()

src/laptop_price/api.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from time import perf_counter
 from uuid import uuid4
 
@@ -11,8 +11,8 @@
 from .logger import get_logger
 from .monitoring import append_prediction_log, read_recent_prediction_logs, summarize_prediction_logs
 from .predict import load_metadata, load_model, predict_price
-from .security import authorize_request
 from .schemas import PredictionLogEntry, PredictionRequest, PredictionResponse
+from .security import authorize_request
 
 logger = get_logger(__name__)
 
@@ -53,7 +53,7 @@ def health() -> dict:
     try:
         metadata = load_metadata()
         trained_at = datetime.fromisoformat(metadata["trained_at_utc"])
-        age_days = (datetime.now(timezone.utc) - trained_at).days
+        age_days = (datetime.now(UTC) - trained_at).days
         result["model_age_days"] = age_days
         if age_days > MODEL_MAX_AGE_DAYS:
             result["warning"] = f"Model is {age_days} days old (threshold: {MODEL_MAX_AGE_DAYS})"
@@ -88,7 +88,11 @@ def metadata(request: Request, _: None = Depends(authorize_request)) -> dict:
 
 
 @v1.get("/predictions/recent", response_model=list[PredictionLogEntry])
-def recent_predictions(request: Request, limit: int = Query(default=RECENT_PREDICTIONS_LIMIT, ge=1, le=100), _: None = Depends(authorize_request)) -> list[dict]:
+def recent_predictions(
+    request: Request,
+    limit: int = Query(default=RECENT_PREDICTIONS_LIMIT, ge=1, le=100),
+    _: None = Depends(authorize_request),
+) -> list[dict]:
     return read_recent_prediction_logs(limit=limit)
 
 
@@ -151,7 +155,11 @@ def metadata_compat(request: Request, _: None = Depends(authorize_request)) -> d
 
 
 @app.get("/predictions/recent", response_model=list[PredictionLogEntry])
-def recent_predictions_compat(request: Request, limit: int = Query(default=RECENT_PREDICTIONS_LIMIT, ge=1, le=100), _: None = Depends(authorize_request)) -> list[dict]:
+def recent_predictions_compat(
+    request: Request,
+    limit: int = Query(default=RECENT_PREDICTIONS_LIMIT, ge=1, le=100),
+    _: None = Depends(authorize_request),
+) -> list[dict]:
     return recent_predictions(request, limit, _)
 
 
@@ -166,5 +174,9 @@ def monitoring_summary_compat(request: Request, _: None = Depends(authorize_requ
 
 
 @app.post("/predict", response_model=PredictionResponse)
-def predict_compat(request: Request, payload: PredictionRequest, _: None = Depends(authorize_request)) -> PredictionResponse:
+def predict_compat(
+    request: Request,
+    payload: PredictionRequest,
+    _: None = Depends(authorize_request),
+) -> PredictionResponse:
     return predict(request, payload, _)

src/laptop_price/config.py

@@ -1,6 +1,7 @@
 import os
 from pathlib import Path
 
+
 def _resolve_root_dir() -> Path:
     env_root = os.getenv("LAPTOP_PRICE_ROOT")
     if env_root:

src/laptop_price/drift.py

@@ -2,15 +2,15 @@
 
 import argparse
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 import pandas as pd
 
 from .config import (
     CATEGORICAL_COLUMNS,
-    DRIFT_ANALYSIS_LIMIT,
     CATEGORICAL_UNSEEN_RATE_THRESHOLD,
+    DRIFT_ANALYSIS_LIMIT,
     FEATURE_COLUMNS,
     LATEST_DRIFT_REPORT_PATH,
     NUMERIC_COLUMNS,
@@ -21,7 +21,6 @@
 from .monitoring import read_recent_prediction_logs
 from .predict import load_metadata
 
-
 UI_OPTION_KEYS = {
     "Company": "companies",
     "TypeName": "types",
@@ -130,7 +129,7 @@ def generate_drift_report(limit: int | None = None) -> dict[str, Any]:
     drift_summary = analyze_feature_drift(inference_df, reference_profile, metadata)
 
     report = {
-        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+        "generated_at_utc": datetime.now(UTC).isoformat(),
         "model_name": metadata["model_name"],
         "model_version": metadata["model_version"],
         "sample_size": int(len(inference_df)),
@@ -152,10 +151,15 @@ def load_latest_drift_report() -> dict[str, Any]:
 
 def main() -> None:
     parser = argparse.ArgumentParser(description="Analyze prediction logs for feature drift.")
-    parser.add_argument("--limit", type=int, default=DRIFT_ANALYSIS_LIMIT, help="Number of recent predictions to analyze")
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=DRIFT_ANALYSIS_LIMIT,
+        help="Number of recent predictions to analyze",
+    )
     args = parser.parse_args()
     print(json.dumps(generate_drift_report(limit=args.limit), indent=2))
 
 
 if __name__ == "__main__":
-    main()
+    main()

src/laptop_price/features.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import asdict, is_dataclass
-from typing import Any, Iterable
+from typing import Any
 
 import numpy as np
 import pandas as pd
@@ -51,10 +52,22 @@ def _parse_memory_components(memory_series: pd.Series) -> pd.DataFrame:
     second_numeric = second.str.replace(r"\D", "", regex=True).replace("", "0").astype(int)
 
     frame = pd.DataFrame(index=memory_series.index)
-    frame["HDD"] = first_numeric * first.str.contains("HDD").astype(int) + second_numeric * second.str.contains("HDD").astype(int)
-    frame["SSD"] = first_numeric * first.str.contains("SSD").astype(int) + second_numeric * second.str.contains("SSD").astype(int)
-    frame["Hybrid"] = first_numeric * first.str.contains("Hybrid").astype(int) + second_numeric * second.str.contains("Hybrid").astype(int)
-    frame["Flash_Storage"] = first_numeric * first.str.contains("Flash Storage").astype(int) + second_numeric * second.str.contains("Flash Storage").astype(int)
+    frame["HDD"] = (
+        first_numeric * first.str.contains("HDD").astype(int)
+        + second_numeric * second.str.contains("HDD").astype(int)
+    )
+    frame["SSD"] = (
+        first_numeric * first.str.contains("SSD").astype(int)
+        + second_numeric * second.str.contains("SSD").astype(int)
+    )
+    frame["Hybrid"] = (
+        first_numeric * first.str.contains("Hybrid").astype(int)
+        + second_numeric * second.str.contains("Hybrid").astype(int)
+    )
+    frame["Flash_Storage"] = (
+        first_numeric * first.str.contains("Flash Storage").astype(int)
+        + second_numeric * second.str.contains("Flash Storage").astype(int)
+    )
     return frame
 
 

src/laptop_price/logger.py

@@ -6,10 +6,9 @@
 import logging
 import os
 import sys
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
-
 LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
 
 
@@ -18,7 +17,7 @@ class JSONFormatter(logging.Formatter):
 
     def format(self, record: logging.LogRecord) -> str:
         log_entry: dict[str, Any] = {
-            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "timestamp": datetime.now(UTC).isoformat(),
             "level": record.levelname,
             "logger": record.name,
             "message": record.getMessage(),

src/laptop_price/monitoring.py

@@ -2,7 +2,7 @@
 
 import json
 import sqlite3
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 from .config import PREDICTION_DB_PATH, PREDICTION_LOG_PATH, RECENT_PREDICTIONS_LIMIT, ensure_directories
@@ -30,7 +30,7 @@ def ensure_prediction_store() -> None:
 def append_prediction_log(payload: dict[str, Any]) -> None:
     ensure_prediction_store()
     log_record = {
-        "logged_at_utc": datetime.now(timezone.utc).isoformat(),
+        "logged_at_utc": datetime.now(UTC).isoformat(),
         **payload,
     }
     with PREDICTION_LOG_PATH.open("a", encoding="utf-8") as handle:
@@ -69,7 +69,8 @@ def read_recent_prediction_logs(limit: int | None = None) -> list[dict[str, Any]
         with sqlite3.connect(PREDICTION_DB_PATH) as connection:
             rows = connection.execute(
                 """
-                SELECT logged_at_utc, request_id, model_name, model_version, predicted_price_inr, latency_ms, features_json
+                SELECT logged_at_utc, request_id, model_name, model_version,
+                       predicted_price_inr, latency_ms, features_json
                 FROM predictions
                 ORDER BY logged_at_utc DESC
                 LIMIT ?
@@ -117,4 +118,4 @@ def summarize_prediction_logs() -> dict[str, Any]:
         "latest_request_id": recent_logs[0]["request_id"],
         "active_model_versions": active_model_versions,
         "average_latency_ms": round(average_latency, 2),
-    }
+    }

src/laptop_price/ops.py

@@ -128,4 +128,4 @@ def main() -> None:
 
 
 if __name__ == "__main__":
-    main()
+    main()

src/laptop_price/performance_history.py

@@ -7,8 +7,7 @@
 from __future__ import annotations
 
 import json
-from datetime import datetime, timezone
-from pathlib import Path
+from datetime import UTC, datetime
 from typing import Any
 
 from .config import METRICS_DIR
@@ -27,7 +26,7 @@ def append_training_run(metadata: dict[str, Any]) -> None:
     """Append a training run entry to the performance history."""
     history = load_performance_history()
     entry = {
-        "recorded_at_utc": datetime.now(timezone.utc).isoformat(),
+        "recorded_at_utc": datetime.now(UTC).isoformat(),
         "model_version": metadata["model_version"],
         "model_name": metadata["model_name"],
         "metrics": metadata["metrics"],

src/laptop_price/security.py

@@ -52,4 +52,4 @@ def enforce_rate_limit(request: Request) -> None:
         if len(request_times) >= RATE_LIMIT_REQUESTS:
             raise HTTPException(status_code=429, detail="Rate limit exceeded")
 
-        request_times.append(now)
+        request_times.append(now)

src/laptop_price/train.py

@@ -4,7 +4,7 @@
 import json
 import shutil
 import tempfile
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -24,8 +24,8 @@
     LATEST_METRICS_PATH,
     METADATA_PATH,
     MODEL_DIR,
-    MODEL_REGISTRY_DIR,
     MODEL_PATH,
+    MODEL_REGISTRY_DIR,
     NUMERIC_COLUMNS,
     RANDOM_STATE,
     REGISTRY_INDEX_PATH,
@@ -238,7 +238,7 @@ def extract_feature_importances(pipeline) -> dict[str, float] | None:
     except AttributeError:
         return None
     importances = model.feature_importances_.tolist()
-    return dict(sorted(zip(feature_names, importances), key=lambda x: x[1], reverse=True))
+    return dict(sorted(zip(feature_names, importances, strict=False), key=lambda x: x[1], reverse=True))
 
 
 def select_best_model(X_train, X_test, y_train, y_test, candidate_models=None, enable_tuning: bool = True):
@@ -274,7 +274,12 @@ def select_best_model(X_train, X_test, y_train, y_test, candidate_models=None, e
             )
             grid_search.fit(X_train, y_train)
             pipeline = grid_search.best_estimator_
-            logger.info("Best params for %s: %s (cv_r2=%.4f)", model_name, grid_search.best_params_, grid_search.best_score_)
+            logger.info(
+                "Best params for %s: %s (cv_r2=%.4f)",
+                model_name,
+                grid_search.best_params_,
+                grid_search.best_score_,
+            )
         else:
             pipeline.fit(X_train, y_train)
 
@@ -294,7 +299,12 @@ def select_best_model(X_train, X_test, y_train, y_test, candidate_models=None, e
     return best_name, best_pipeline, best_metrics, all_metrics, cv_scores_by_model
 
 
-def train_and_save(raw_data_path: str | None = None, model_dir: Path | None = None, metrics_path: Path | None = None, enable_tuning: bool = True):
+def train_and_save(
+    raw_data_path: str | None = None,
+    model_dir: Path | None = None,
+    metrics_path: Path | None = None,
+    enable_tuning: bool = True,
+):
     ensure_directories()
     logger.info("Starting training pipeline")
 
@@ -326,8 +336,12 @@ def train_and_save(raw_data_path: str | None = None, model_dir: Path | None = No
 
     target_metrics_path = metrics_path or LATEST_METRICS_PATH
 
-    trained_at = datetime.now(timezone.utc)
-    resolved_data_path = Path(raw_data_path).resolve() if raw_data_path is not None else resolve_raw_data_path().resolve()
+    trained_at = datetime.now(UTC)
+    resolved_data_path = (
+        Path(raw_data_path).resolve()
+        if raw_data_path is not None
+        else resolve_raw_data_path().resolve()
+    )
     model_version = trained_at.strftime("%Y%m%d%H%M%S%f")
 
     # Extract feature importances from final pipeline
@@ -390,8 +404,17 @@ def main() -> None:
     parser = argparse.ArgumentParser(description="Train and persist the laptop price model.")
     parser.add_argument("--data", dest="data_path", default=None, help="Optional path to laptop_data.csv")
     parser.add_argument("--list-versions", action="store_true", help="List registered model versions")
-    parser.add_argument("--activate-version", dest="activate_version", default=None, help="Promote a registered model version to production")
-    parser.add_argument("--no-tuning", action="store_true", help="Skip GridSearchCV hyperparameter tuning for faster training")
+    parser.add_argument(
+        "--activate-version",
+        dest="activate_version",
+        default=None,
+        help="Promote a registered model version to production",
+    )
+    parser.add_argument(
+        "--no-tuning",
+        action="store_true",
+        help="Skip GridSearchCV hyperparameter tuning for faster training",
+    )
     args = parser.parse_args()
 
     if args.list_versions:

src/laptop_price/validation.py

@@ -5,8 +5,7 @@
 import pandera.pandas as pa
 from pandera.pandas import Column, DataFrameSchema
 
-from .config import CATEGORICAL_COLUMNS, FEATURE_COLUMNS, NUMERIC_COLUMNS, TARGET_COLUMN
-
+from .config import TARGET_COLUMN
 
 training_schema = DataFrameSchema(
     columns={

tests/conftest.py

@@ -1,7 +1,7 @@
-from pathlib import Path
 import sys
+from pathlib import Path
 
 ROOT_DIR = Path(__file__).resolve().parents[1]
 SRC_DIR = ROOT_DIR / "src"
 if str(SRC_DIR) not in sys.path:
-    sys.path.insert(0, str(SRC_DIR))
+    sys.path.insert(0, str(SRC_DIR))

tests/test_alerts.py

@@ -12,4 +12,4 @@ def test_evaluate_drift_alert_fails_when_drift_detected_with_enough_samples() ->
     evaluation = evaluate_drift_alert({"sample_size": 50, "drift_detected": True})
 
     assert evaluation["should_alert"] is True
-    assert evaluation["reason"] == "drift_detected"
+    assert evaluation["reason"] == "drift_detected"