Coverage for mindsdb / integrations / utilities / date_utils.py: 0%

1import datetime as dt 

2import pytz 

3import re 

4 

5 

6def parse_local_date(date_str: str) -> dt.datetime: 

7 """Parses common date string formats to local datetime objects.""" 

8 if isinstance(date_str, dt.datetime): 

9 return date_str 

10 date_formats = ['%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d'] 

11 

12 date = None 

13 for date_format in date_formats: 

14 try: 

15 date = dt.datetime.strptime(date_str, date_format) 

16 except ValueError: 

17 pass 

18 if date is None: 

19 raise ValueError(f"Can't parse date: {date_str}") 

20 return date 

21 

22 

23def parse_utc_date_with_limit(date_str: str, max_window_in_days: int = None) -> dt.datetime: 

24 """Parses common date string formats to UTC datetime objects.""" 

25 date = parse_local_date(date_str) 

26 

27 # Convert date to UTC 

28 date_utc = date.astimezone(pytz.utc) 

29 

30 # If max_window_in_days is provided, apply the logic 

31 if max_window_in_days is not None: 

32 # Get the current UTC time 

33 now_utc = dt.datetime.utcnow().replace(tzinfo=pytz.utc) 

34 # Check if the parsed date is earlier than the maximum window allowed 

35 max_window_date = now_utc - dt.timedelta(days=max_window_in_days) 

36 if date_utc < max_window_date: 

37 return max_window_date 

38 return date_utc 

39 

40 

41def parse_utc_date(date_str: str) -> dt.datetime: 

42 """Parses common date string formats to UTC datetime objects.""" 

43 date = parse_local_date(date_str) 

44 return date.astimezone(pytz.utc) 

45 

46 

47def utc_date_str_to_timestamp_ms(date_str: str) -> int: 

48 """Parses common date string formats into ms since the Unix epoch in UTC.""" 

49 date = parse_local_date(date_str) 

50 # `timestamp` method doesn't work as expected unless we replace the timezone info this way. 

51 date = date.replace(tzinfo=pytz.UTC) 

52 return int(date.timestamp() * 1000) 

53 

54 

55def interval_str_to_duration_ms(interval_str: str) -> int: 

56 """Parses interval strings into how long they represent in ms. 

57 Supported intervals: 

58 - seconds (e.g. 1s) 

59 - minutes (e.g. 5m) 

60 - hours (e.g. 1h) 

61 - days (e.g. 5d) 

62 - weeks (e.g. 1w) 

63 """ 

64 duration_match = re.search(r'^\d+', interval_str) 

65 time_unit_match = re.search('[smhdw]', interval_str) 

66 if not duration_match or not time_unit_match: 

67 raise ValueError('Invalid interval {}'.format(interval_str)) 

68 duration = int(duration_match.group()) 

69 time_unit = time_unit_match.group() 

70 if time_unit == 's': 

71 return duration * 1000 

72 if time_unit == 'm': 

73 return duration * 1000 * 60 

74 if time_unit == 'h': 

75 return duration * 1000 * 60 * 60 

76 if time_unit == 'd': 

77 return duration * 1000 * 60 * 60 * 24 

78 if time_unit == 'w': 

79 return duration * 1000 * 60 * 60 * 24 * 7