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
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1from typing import Optional
3import schemas
4from chalicelib.core.sourcemaps import sourcemaps
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
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
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