mirror of
https://github.com/PostHog/posthog.git
synced 2024-12-01 04:12:23 +01:00
55d5c6faa0
* Use GroupTypeIndex type in BE * Consistent ordering for group types * Better BE type for group type index * Handle non-url safe keys in groups Previously this would crash due to never decoding the url and not handling it in the backend * Allow fetching related groups from the API * Solve minor type errors * Add a tooltip * data-attr for tracking links * Add related groups onto person page * Fix an import * Kill useEffect
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
import dataclasses
|
|
from typing import Any, Dict, Literal, Optional, TypedDict, Union
|
|
|
|
from django.test.client import Client
|
|
|
|
from ee.clickhouse.queries.funnels.funnel_correlation import EventOddsRatioSerialized
|
|
from posthog.constants import FunnelCorrelationType
|
|
from posthog.models.property import GroupTypeIndex
|
|
|
|
|
|
class EventPattern(TypedDict, total=False):
|
|
id: str
|
|
type: Union[Literal["events"], Literal["actions"]]
|
|
order: int
|
|
properties: Dict[str, Any]
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class FunnelCorrelationRequest:
|
|
# Needs to be json encoded list of `EventPattern`s
|
|
events: str
|
|
date_to: str
|
|
funnel_step: Optional[int] = None
|
|
date_from: Optional[str] = None
|
|
funnel_correlation_type: Optional[FunnelCorrelationType] = None
|
|
# Needs to be json encoded list of `str`s
|
|
funnel_correlation_names: Optional[str] = None
|
|
funnel_correlation_event_names: Optional[str] = None
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class FunnelRequest:
|
|
events: str
|
|
date_from: str
|
|
insight: str
|
|
aggregation_group_type_index: Optional[GroupTypeIndex] = None
|
|
date_to: Optional[str] = None
|
|
properties: Optional[str] = None
|
|
funnel_order_type: Optional[str] = None
|
|
|
|
|
|
def get_funnel(client: Client, team_id: int, request: FunnelRequest):
|
|
return client.post(
|
|
f"/api/projects/{team_id}/insights/funnel",
|
|
data={key: value for key, value in dataclasses.asdict(request).items() if value is not None},
|
|
)
|
|
|
|
|
|
def get_funnel_ok(client: Client, team_id: int, request: FunnelRequest) -> Dict[str, Any]:
|
|
response = get_funnel(client=client, team_id=team_id, request=request)
|
|
|
|
assert response.status_code == 200, response.content
|
|
res = response.json()
|
|
final = {}
|
|
|
|
for step in res["result"]:
|
|
final[step["name"]] = step
|
|
|
|
return final
|
|
|
|
|
|
def get_funnel_actors_ok(client: Client, url: str):
|
|
response = client.get(url)
|
|
|
|
assert response.status_code == 200, response.content
|
|
return response.json()["results"][0]["people"]
|
|
|
|
|
|
def get_funnel_correlation(client: Client, team_id: int, request: FunnelCorrelationRequest):
|
|
return client.get(
|
|
f"/api/projects/{team_id}/insights/funnel/correlation",
|
|
data={key: value for key, value in dataclasses.asdict(request).items() if value is not None},
|
|
)
|
|
|
|
|
|
def get_funnel_correlation_ok(client: Client, team_id: int, request: FunnelCorrelationRequest) -> Dict[str, Any]:
|
|
response = get_funnel_correlation(client=client, team_id=team_id, request=request)
|
|
|
|
assert response.status_code == 200, response.content
|
|
return response.json()
|
|
|
|
|
|
def get_people_for_correlation_ok(client: Client, correlation: EventOddsRatioSerialized) -> Dict[str, Any]:
|
|
"""
|
|
Helper for getting people for a correlation. Note we keep checking to just
|
|
inclusion of name, to make the stable to changes in other people props.
|
|
"""
|
|
success_people_url = correlation["success_people_url"]
|
|
failure_people_url = correlation["failure_people_url"]
|
|
|
|
if not success_people_url or not failure_people_url:
|
|
return {}
|
|
|
|
success_people_response = client.get(success_people_url)
|
|
assert success_people_response.status_code == 200, success_people_response.content
|
|
|
|
failure_people_response = client.get(failure_people_url)
|
|
assert failure_people_response.status_code == 200, failure_people_response.content
|
|
|
|
return {
|
|
"success": sorted([person["name"] for person in success_people_response.json()["results"][0]["people"]]),
|
|
"failure": sorted([person["name"] for person in failure_people_response.json()["results"][0]["people"]]),
|
|
}
|