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 

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

10 """Get SQLAlchemy condition for time filtering. 

11 

12 Args: 

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

14 timestamp_column: SQLAlchemy timestamp column to filter on 

15 

16 Returns: 

17 SQLAlchemy condition object or None for 'all' 

18 """ 

19 if period == "all": 

20 return None 

21 elif period == "7d": 

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

23 elif period == "30d": 

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

25 elif period == "3m": 

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

27 elif period == "1y": 

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

29 else: 

30 # Default to 30 days for unknown periods 

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

32 

33 return timestamp_column >= cutoff 

34 

35 

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

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

38 

39 Args: 

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

41 mode_column: SQLAlchemy column to filter on 

42 

43 Returns: 

44 SQLAlchemy condition object or None for 'all' 

45 """ 

46 if research_mode == "all": 

47 return None 

48 elif research_mode in ["quick", "detailed"]: 

49 return mode_column == research_mode 

50 else: 

51 return None