Coverage for src / local_deep_research / metrics / query_utils.py: 100%

1"""Common query utilities for metrics module.""" 

2 

3from datetime import datetime, timedelta, UTC 

4from typing import Any 

5 

6from sqlalchemy import Column 

7 

8 

9PERIOD_DAYS_MAP = {"7d": 7, "30d": 30, "90d": 90, "365d": 365, "all": None} 

10 

11 

12def get_period_days(period: str, default: int = 30) -> int | None: 

13 """Convert a period string to number of days. 

14 

15 Returns None for 'all' (no time limit). 

16 """ 

17 return PERIOD_DAYS_MAP.get(period, default) 

18 

19 

20def get_time_filter_condition(period: str, timestamp_column: Column) -> Any: 

21 """Get SQLAlchemy condition for time filtering. 

22 

23 Args: 

24 period: Time period ('7d', '30d', '3m', '1y', 'all') 

25 timestamp_column: SQLAlchemy timestamp column to filter on 

26 

27 Returns: 

28 SQLAlchemy condition object or None for 'all' 

29 """ 

30 if period == "all": 

31 return None 

32 if period == "7d": 

33 cutoff = datetime.now(UTC) - timedelta(days=7) 

34 elif period == "30d": 

35 cutoff = datetime.now(UTC) - timedelta(days=30) 

36 elif period == "3m": 

37 cutoff = datetime.now(UTC) - timedelta(days=90) 

38 elif period == "1y": 

39 cutoff = datetime.now(UTC) - timedelta(days=365) 

40 else: 

41 # Default to 30 days for unknown periods 

42 cutoff = datetime.now(UTC) - timedelta(days=30) 

43 

44 return timestamp_column >= cutoff 

45 

46 

47def get_research_mode_condition(research_mode: str, mode_column: Column) -> Any: 

48 """Get SQLAlchemy condition for research mode filtering. 

49 

50 Args: 

51 research_mode: Research mode ('quick', 'detailed', 'all') 

52 mode_column: SQLAlchemy column to filter on 

53 

54 Returns: 

55 SQLAlchemy condition object or None for 'all' 

56 """ 

57 if research_mode == "all": 

58 return None 

59 if research_mode in ["quick", "detailed"]: 

60 return mode_column == research_mode 

61 return None