jan/openreplay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(chalice): refactored code

Browse source
This commit is contained in:
Taha Yassine Kraiem 2025-03-27 12:48:45 +01:00 • committed by Kraiem Taha Yassine
parent e0dba41065
commit 9db5e2a8f7
9 changed files with 33 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api/chalicelib/core/metrics/funnels.py
View file

@ -6,7 +6,7 @@ from chalicelib.utils import helper
from chalicelib.utils import sql_helper as sh
def filter_stages(stages: List[schemas.SessionSearchEventSchema2]):
def filter_stages(stages: List[schemas.SessionSearchEventSchema]):
ALLOW_TYPES = [schemas.EventType.CLICK, schemas.EventType.INPUT,
schemas.EventType.LOCATION, schemas.EventType.CUSTOM,
schemas.EventType.CLICK_MOBILE, schemas.EventType.INPUT_MOBILE,
@ -15,10 +15,10 @@ def filter_stages(stages: List[schemas.SessionSearchEventSchema2]):
def __parse_events(f_events: List[dict]):
return [schemas.SessionSearchEventSchema2.parse_obj(e) for e in f_events]
return [schemas.SessionSearchEventSchema.parse_obj(e) for e in f_events]
def __fix_stages(f_events: List[schemas.SessionSearchEventSchema2]):
def __fix_stages(f_events: List[schemas.SessionSearchEventSchema]):
if f_events is None:
return
events = []

6
api/chalicelib/core/metrics/heatmaps/heatmaps.py
View file

@ -160,7 +160,7 @@ s.start_ts,
s.duration"""
def __get_1_url(location_condition: schemas.SessionSearchEventSchema2 | None, session_id: str, project_id: int,
def __get_1_url(location_condition: schemas.SessionSearchEventSchema | None, session_id: str, project_id: int,
start_time: int,
end_time: int) -> str | None:
full_args = {
@ -240,11 +240,11 @@ def search_short_session(data: schemas.HeatMapSessionsSearch, project_id, user_i
value=[schemas.PlatformType.DESKTOP],
operator=schemas.SearchEventOperator.IS))
if not location_condition:
data.events.append(schemas.SessionSearchEventSchema2(type=schemas.EventType.LOCATION,
data.events.append(schemas.SessionSearchEventSchema(type=schemas.EventType.LOCATION,
value=[],
operator=schemas.SearchEventOperator.IS_ANY))
if no_click:
data.events.append(schemas.SessionSearchEventSchema2(type=schemas.EventType.CLICK,
data.events.append(schemas.SessionSearchEventSchema(type=schemas.EventType.CLICK,
value=[],
operator=schemas.SearchEventOperator.IS_ANY))

6
api/chalicelib/core/metrics/heatmaps/heatmaps_ch.py
View file

@ -179,7 +179,7 @@ toUnixTimestamp(s.datetime)*1000 AS start_ts,
s.duration AS duration"""
def __get_1_url(location_condition: schemas.SessionSearchEventSchema2 | None, session_id: str, project_id: int,
def __get_1_url(location_condition: schemas.SessionSearchEventSchema | None, session_id: str, project_id: int,
start_time: int,
end_time: int) -> str | None:
full_args = {
@ -262,11 +262,11 @@ def search_short_session(data: schemas.HeatMapSessionsSearch, project_id, user_i
value=[schemas.PlatformType.DESKTOP],
operator=schemas.SearchEventOperator.IS))
if not location_condition:
data.events.append(schemas.SessionSearchEventSchema2(type=schemas.EventType.LOCATION,
data.events.append(schemas.SessionSearchEventSchema(type=schemas.EventType.LOCATION,
value=[],
operator=schemas.SearchEventOperator.IS_ANY))
if no_click:
data.events.append(schemas.SessionSearchEventSchema2(type=schemas.EventType.CLICK,
data.events.append(schemas.SessionSearchEventSchema(type=schemas.EventType.CLICK,
value=[],
operator=schemas.SearchEventOperator.IS_ANY))

2
api/chalicelib/core/metrics/modules/significance/significance.py
View file

@ -241,7 +241,7 @@ def get_simple_funnel(filter_d: schemas.CardSeriesFilterSchema, project: schemas
:return:
"""
stages: List[schemas.SessionSearchEventSchema2] = filter_d.events
stages: List[schemas.SessionSearchEventSchema] = filter_d.events
filters: List[schemas.SessionSearchFilterSchema] = filter_d.filters
stage_constraints = ["main.timestamp <= %(endTimestamp)s"]

2
api/chalicelib/core/metrics/modules/significance/significance_ch.py
View file

@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
def get_simple_funnel(filter_d: schemas.CardSeriesFilterSchema, project: schemas.ProjectContext,
metric_format: schemas.MetricExtendedFormatType) -> List[RealDictRow]:
stages: List[schemas.SessionSearchEventSchema2] = filter_d.events
stages: List[schemas.SessionSearchEventSchema] = filter_d.events
filters: List[schemas.SessionSearchFilterSchema] = filter_d.filters
platform = project.platform
constraints = ["e.project_id = %(project_id)s",

6
api/chalicelib/core/sessions/sessions_ch.py
View file

@ -148,7 +148,7 @@ def search2_table(data: schemas.SessionsSearchPayloadSchema, project_id: int, de
for e in data.events:
if e.type == schemas.EventType.LOCATION:
if e.operator not in extra_conditions:
extra_conditions[e.operator] = schemas.SessionSearchEventSchema2.model_validate({
extra_conditions[e.operator] = schemas.SessionSearchEventSchema.model_validate({
"type": e.type,
"isEvent": True,
"value": [],
@ -173,7 +173,7 @@ def search2_table(data: schemas.SessionsSearchPayloadSchema, project_id: int, de
for e in data.events:
if e.type == schemas.EventType.REQUEST_DETAILS:
if e.operator not in extra_conditions:
extra_conditions[e.operator] = schemas.SessionSearchEventSchema2.model_validate({
extra_conditions[e.operator] = schemas.SessionSearchEventSchema.model_validate({
"type": e.type,
"isEvent": True,
"value": [],
@ -277,7 +277,7 @@ def search2_table(data: schemas.SessionsSearchPayloadSchema, project_id: int, de
return sessions
def __is_valid_event(is_any: bool, event: schemas.SessionSearchEventSchema2):
def __is_valid_event(is_any: bool, event: schemas.SessionSearchEventSchema):
return not (not is_any and len(event.value) == 0 and event.type not in [schemas.EventType.REQUEST_DETAILS,
schemas.EventType.GRAPHQL] \
or event.type in [schemas.PerformanceEventType.LOCATION_DOM_COMPLETE,

2
api/chalicelib/core/sessions/sessions_legacy_mobil.py
View file

@ -218,7 +218,7 @@ def search_sessions(data: schemas.SessionsSearchPayloadSchema, project_id, user_
}
def __is_valid_event(is_any: bool, event: schemas.SessionSearchEventSchema2):
def __is_valid_event(is_any: bool, event: schemas.SessionSearchEventSchema):
return not (not is_any and len(event.value) == 0 and event.type not in [schemas.EventType.REQUEST_DETAILS,
schemas.EventType.GRAPHQL] \
or event.type in [schemas.PerformanceEventType.LOCATION_DOM_COMPLETE,

6
api/chalicelib/core/sessions/sessions_pg.py
View file

@ -143,7 +143,7 @@ def search2_table(data: schemas.SessionsSearchPayloadSchema, project_id: int, de
for e in data.events:
if e.type == schemas.EventType.LOCATION:
if e.operator not in extra_conditions:
extra_conditions[e.operator] = schemas.SessionSearchEventSchema2.model_validate({
extra_conditions[e.operator] = schemas.SessionSearchEventSchema.model_validate({
"type": e.type,
"isEvent": True,
"value": [],
@ -160,7 +160,7 @@ def search2_table(data: schemas.SessionsSearchPayloadSchema, project_id: int, de
for e in data.events:
if e.type == schemas.EventType.REQUEST_DETAILS:
if e.operator not in extra_conditions:
extra_conditions[e.operator] = schemas.SessionSearchEventSchema2.model_validate({
extra_conditions[e.operator] = schemas.SessionSearchEventSchema.model_validate({
"type": e.type,
"isEvent": True,
"value": [],
@ -273,7 +273,7 @@ def search2_table(data: schemas.SessionsSearchPayloadSchema, project_id: int, de
return sessions
def __is_valid_event(is_any: bool, event: schemas.SessionSearchEventSchema2):
def __is_valid_event(is_any: bool, event: schemas.SessionSearchEventSchema):
return not (not is_any and len(event.value) == 0 and event.type not in [schemas.EventType.REQUEST_DETAILS,
schemas.EventType.GRAPHQL] \
or event.type in [schemas.PerformanceEventType.LOCATION_DOM_COMPLETE,

11
api/schemas/schemas.py
View file

@ -584,6 +584,7 @@ class PropertyFilterSchema(BaseModel):
name: Union[EventPredefinedPropertyType, str] = Field(...)
operator: Union[SearchEventOperator, MathOperator] = Field(...)
value: List[Union[int, str]] = Field(...)
# property_type: Optional[Literal["string", "number", "date"]] = Field(default=None)
@computed_field
@ -603,7 +604,7 @@ class EventPropertiesSchema(BaseModel):
filters: List[PropertyFilterSchema] = Field(...)
class SessionSearchEventSchema2(BaseModel):
class SessionSearchEventSchema(BaseModel):
is_event: Literal[True] = True
value: List[Union[str, int]] = Field(...)
type: Union[EventType, PerformanceEventType] = Field(...)
@ -719,12 +720,12 @@ def add_missing_is_event(values: dict):
# this type is created to allow mixing events&filters and specifying a discriminator
GroupedFilterType = Annotated[Union[SessionSearchFilterSchema, SessionSearchEventSchema2],
GroupedFilterType = Annotated[Union[SessionSearchFilterSchema, SessionSearchEventSchema],
Field(discriminator='is_event'), BeforeValidator(add_missing_is_event)]
class SessionsSearchPayloadSchema(_TimedSchema, _PaginatedSchema):
events: List[SessionSearchEventSchema2] = Field(default_factory=list, doc_hidden=True)
events: List[SessionSearchEventSchema] = Field(default_factory=list, doc_hidden=True)
filters: List[GroupedFilterType] = Field(default_factory=list)
sort: str = Field(default="startTs")
order: SortOrderType = Field(default=SortOrderType.DESC)
@ -749,6 +750,8 @@ class SessionsSearchPayloadSchema(_TimedSchema, _PaginatedSchema):
def add_missing_attributes(cls, values):
# in case isEvent is wrong:
for f in values.get("filters") or []:
if f.get("type") is None:
continue
if EventType.has_value(f["type"]) and not f.get("isEvent"):
f["isEvent"] = True
elif FilterType.has_value(f["type"]) and f.get("isEvent"):
@ -1463,7 +1466,7 @@ class MetricSearchSchema(_PaginatedSchema):
mine_only: bool = Field(default=False)
class _HeatMapSearchEventRaw(SessionSearchEventSchema2):
class _HeatMapSearchEventRaw(SessionSearchEventSchema):
type: Literal[EventType.LOCATION] = Field(...)