Coverage for chalicelib/core/errors/helper.py: 27%

40 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 12:56 +0000

1from typing import Optional 

2 

3import schemas 

4from chalicelib.core.sourcemaps import sourcemaps 

5 

6 

7def __get_basic_constraints(platform: Optional[schemas.PlatformType] = None, time_constraint: bool = True, 

8 startTime_arg_name: str = "startDate", endTime_arg_name: str = "endDate", 

9 chart: bool = False, step_size_name: str = "step_size", 

10 project_key: Optional[str] = "project_id"): 

11 if project_key is None: 

12 ch_sub_query = [] 

13 else: 

14 ch_sub_query = [f"{project_key} =%(project_id)s"] 

15 if time_constraint: 

16 ch_sub_query += [f"timestamp >= %({startTime_arg_name})s", 

17 f"timestamp < %({endTime_arg_name})s"] 

18 if chart: 

19 ch_sub_query += [f"timestamp >= generated_timestamp", 

20 f"timestamp < generated_timestamp + %({step_size_name})s"] 

21 if platform == schemas.PlatformType.MOBILE: 

22 ch_sub_query.append("user_device_type = 'mobile'") 

23 elif platform == schemas.PlatformType.DESKTOP: 

24 ch_sub_query.append("user_device_type = 'desktop'") 

25 return ch_sub_query 

26 

27 

28def __get_basic_constraints_ch(platform=None, time_constraint=True, startTime_arg_name="startDate", 

29 endTime_arg_name="endDate", type_condition=True, project_key="project_id", 

30 table_name=None): 

31 ch_sub_query = [f"{project_key} =toUInt16(%(project_id)s)"] 

32 if table_name is not None: 32 ↛ 35line 32 didn't jump to line 35 because the condition on line 32 was always true

33 table_name = table_name + "." 

34 else: 

35 table_name = "" 

36 if type_condition: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 ch_sub_query.append(f"{table_name}`$event_name`='ERROR'") 

38 if time_constraint: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 ch_sub_query += [f"{table_name}datetime >= toDateTime(%({startTime_arg_name})s/1000)", 

40 f"{table_name}datetime < toDateTime(%({endTime_arg_name})s/1000)"] 

41 if platform == schemas.PlatformType.MOBILE: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 ch_sub_query.append("user_device_type = 'mobile'") 

43 elif platform == schemas.PlatformType.DESKTOP: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true

44 ch_sub_query.append("user_device_type = 'desktop'") 

45 return ch_sub_query 

46 

47 

48def format_first_stack_frame(error): 

49 error["stack"] = sourcemaps.format_payload(error.pop("payload"), truncate_to_first=True) 

50 for s in error["stack"]: 

51 for c in s.get("context", []): 

52 for sci, sc in enumerate(c): 

53 if isinstance(sc, str) and len(sc) > 1000: 

54 c[sci] = sc[:1000] 

55 # convert bytes to string: 

56 if isinstance(s["filename"], bytes): 

57 s["filename"] = s["filename"].decode("utf-8") 

58 return error