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

feat(chalice): refactored projects code

Browse source 
feat(chalice): sessions-check-flag every hour
feat(chalice): sessions-check-delta set to 4 hours
This commit is contained in:
Taha Yassine Kraiem 2023-04-26 17:47:00 +02:00
parent b8dfe90ce8
commit 171373b73e
5 changed files with 31 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
api/chalicelib/core/projects.py
View file

@ -58,21 +58,22 @@ def get_projects(tenant_id: int, gdpr: bool = False, recorded: bool = False):
if gdpr: if gdpr:
extra_projection += ',s.gdpr' extra_projection += ',s.gdpr'
if recorded: if recorded:
extra_projection += """,COALESCE(EXTRACT(EPOCH FROM s.first_recorded_session_at) * 1000::BIGINT, extra_projection += """,\nCOALESCE(EXTRACT(EPOCH FROM s.first_recorded_session_at) * 1000::BIGINT,
(SELECT MIN(sessions.start_ts) (SELECT MIN(sessions.start_ts)
FROM public.sessions FROM public.sessions
WHERE sessions.project_id = s.project_id WHERE sessions.project_id = s.project_id
AND sessions.start_ts >= (EXTRACT(EPOCH FROM AND sessions.start_ts >= (EXTRACT(EPOCH
COALESCE(s.sessions_last_check_at, s.created_at)) * 1000-24*60*60*1000) FROM COALESCE(s.sessions_last_check_at, s.created_at)) * 1000-%(check_delta)s)
AND sessions.start_ts <= %(now)s AND sessions.start_ts <= %(now)s
)) AS first_recorded""" )) AS first_recorded"""
query = cur.mogrify(f"""{"SELECT *, first_recorded IS NOT NULL AS recorded FROM (" if recorded else ""} query = cur.mogrify(f"""{"SELECT *, first_recorded IS NOT NULL AS recorded FROM (" if recorded else ""}
SELECT s.project_id, s.name, s.project_key, s.save_request_payloads, s.first_recorded_session_at, SELECT s.project_id, s.name, s.project_key, s.save_request_payloads, s.first_recorded_session_at,
created_at {extra_projection} created_at, sessions_last_check_at {extra_projection}
FROM public.projects AS s FROM public.projects AS s
WHERE s.deleted_at IS NULL WHERE s.deleted_at IS NULL
ORDER BY s.name {") AS raw" if recorded else ""};""", {"now": TimeUTC.now()}) ORDER BY s.name {") AS raw" if recorded else ""};""",
{"now": TimeUTC.now(), "check_delta": TimeUTC.MS_HOUR * 4})
cur.execute(query) cur.execute(query)
rows = cur.fetchall() rows = cur.fetchall()
# if recorded is requested, check if it was saved or computed # if recorded is requested, check if it was saved or computed
@ -80,13 +81,17 @@ def get_projects(tenant_id: int, gdpr: bool = False, recorded: bool = False):
u_values = [] u_values = []
params = {} params = {}
for i, r in enumerate(rows): for i, r in enumerate(rows):
r["sessions_last_check_at"] = TimeUTC.datetime_to_timestamp(r["sessions_last_check_at"])
r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"]) r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"])
if r["first_recorded_session_at"] is None: if r["first_recorded_session_at"] is None \
and r["sessions_last_check_at"] is not None \
and (TimeUTC.now() - r["sessions_last_check_at"]) > TimeUTC.MS_HOUR:
u_values.append(f"(%(project_id_{i})s,to_timestamp(%(first_recorded_{i})s/1000))") u_values.append(f"(%(project_id_{i})s,to_timestamp(%(first_recorded_{i})s/1000))")
params[f"project_id_{i}"] = r["project_id"] params[f"project_id_{i}"] = r["project_id"]
params[f"first_recorded_{i}"] = r["first_recorded"] if r["recorded"] else None params[f"first_recorded_{i}"] = r["first_recorded"] if r["recorded"] else None
r.pop("first_recorded_session_at") r.pop("first_recorded_session_at")
r.pop("first_recorded") r.pop("first_recorded")
r.pop("sessions_last_check_at")
if len(u_values) > 0: if len(u_values) > 0:
query = cur.mogrify(f"""UPDATE public.projects query = cur.mogrify(f"""UPDATE public.projects
SET sessions_last_check_at=(now() at time zone 'utc'), first_recorded_session_at=u.first_recorded SET sessions_last_check_at=(now() at time zone 'utc'), first_recorded_session_at=u.first_recorded
@ -96,6 +101,7 @@ def get_projects(tenant_id: int, gdpr: bool = False, recorded: bool = False):
else: else:
for r in rows: for r in rows:
r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"]) r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"])
r.pop("sessions_last_check_at")
return helper.list_to_camel_case(rows) return helper.list_to_camel_case(rows)

55
ee/api/chalicelib/core/projects.py
View file

@ -32,7 +32,7 @@ def __update(tenant_id, project_id, changes):
for key in changes.keys(): for key in changes.keys():
sub_query.append(f"{helper.key_to_snake_case(key)} = %({key})s") sub_query.append(f"{helper.key_to_snake_case(key)} = %({key})s")
with pg_client.PostgresClient() as cur: with pg_client.PostgresClient() as cur:
query = cur.mogrify(f"""UPDATE public.projects query = cur.mogrify(f"""UPDATE public.projects
SET {" ,".join(sub_query)} SET {" ,".join(sub_query)}
WHERE project_id = %(project_id)s WHERE project_id = %(project_id)s
AND deleted_at ISNULL AND deleted_at ISNULL
@ -53,8 +53,7 @@ def __create(tenant_id, name):
return get_project(tenant_id=tenant_id, project_id=project_id, include_gdpr=True) return get_project(tenant_id=tenant_id, project_id=project_id, include_gdpr=True)
def get_projects(tenant_id, recording_state=False, gdpr=None, recorded=False, stack_integrations=False, user_id=None): def get_projects(tenant_id: int, gdpr: bool = False, recorded: bool = False, user_id: int = None):
stack_integrations = False
with pg_client.PostgresClient() as cur: with pg_client.PostgresClient() as cur:
role_query = """INNER JOIN LATERAL (SELECT 1 role_query = """INNER JOIN LATERAL (SELECT 1
FROM users FROM users
@ -66,37 +65,28 @@ def get_projects(tenant_id, recording_state=False, gdpr=None, recorded=False, st
AND (roles.all_projects OR roles_projects.project_id = s.project_id) AND (roles.all_projects OR roles_projects.project_id = s.project_id)
LIMIT 1) AS role_project ON (TRUE)""" LIMIT 1) AS role_project ON (TRUE)"""
extra_projection = "" extra_projection = ""
extra_join = ""
if gdpr: if gdpr:
extra_projection += ',s.gdpr' extra_projection += ',s.gdpr'
if recorded: if recorded:
extra_projection += """,COALESCE(nullif(EXTRACT(EPOCH FROM s.first_recorded_session_at) * 1000, NULL)::BIGINT , extra_projection += """,\nCOALESCE(EXTRACT(EPOCH FROM s.first_recorded_session_at) * 1000::BIGINT,
(SELECT MIN(sessions.start_ts) (SELECT MIN(sessions.start_ts)
FROM public.sessions FROM public.sessions
WHERE sessions.project_id = s.project_id WHERE sessions.project_id = s.project_id
AND sessions.start_ts >= (EXTRACT(EPOCH FROM AND sessions.start_ts >= (EXTRACT(EPOCH
COALESCE(s.sessions_last_check_at, s.created_at)) * 1000-24*60*60*1000) FROM COALESCE(s.sessions_last_check_at, s.created_at)) * 1000-%(check_delta)s)
AND sessions.start_ts <= %(now)s AND sessions.start_ts <= %(now)s
LIMIT 1), NULL) AS first_recorded""" )) AS first_recorded"""
if stack_integrations:
extra_projection += ',stack_integrations.count>0 AS stack_integrations'
if stack_integrations:
extra_join = """LEFT JOIN LATERAL (SELECT COUNT(*) AS count
FROM public.integrations
WHERE s.project_id = integrations.project_id
LIMIT 1) AS stack_integrations ON TRUE"""
query = cur.mogrify(f"""{"SELECT *, first_recorded IS NOT NULL AS recorded FROM (" if recorded else ""} query = cur.mogrify(f"""{"SELECT *, first_recorded IS NOT NULL AS recorded FROM (" if recorded else ""}
SELECT s.project_id, s.name, s.project_key, s.save_request_payloads, s.first_recorded_session_at, SELECT s.project_id, s.name, s.project_key, s.save_request_payloads, s.first_recorded_session_at,
created_at {extra_projection} created_at, sessions_last_check_at {extra_projection}
FROM public.projects AS s FROM public.projects AS s
{extra_join}
{role_query if user_id is not None else ""} {role_query if user_id is not None else ""}
WHERE s.tenant_id =%(tenant_id)s WHERE s.tenant_id =%(tenant_id)s
AND s.deleted_at IS NULL AND s.deleted_at IS NULL
ORDER BY s.name {") AS raw" if recorded else ""};""", ORDER BY s.name {") AS raw" if recorded else ""};""",
{"tenant_id": tenant_id, "user_id": user_id, "now": TimeUTC.now()}) {"tenant_id": tenant_id, "user_id": user_id, "now": TimeUTC.now(),
"check_delta": TimeUTC.MS_HOUR * 4})
cur.execute(query) cur.execute(query)
rows = cur.fetchall() rows = cur.fetchall()
# if recorded is requested, check if it was saved or computed # if recorded is requested, check if it was saved or computed
@ -104,13 +94,17 @@ def get_projects(tenant_id, recording_state=False, gdpr=None, recorded=False, st
u_values = [] u_values = []
params = {} params = {}
for i, r in enumerate(rows): for i, r in enumerate(rows):
r["sessions_last_check_at"] = TimeUTC.datetime_to_timestamp(r["sessions_last_check_at"])
r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"]) r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"])
if r["first_recorded_session_at"] is None: if r["first_recorded_session_at"] is None \
and r["sessions_last_check_at"] is not None \
and (TimeUTC.now() - r["sessions_last_check_at"]) > TimeUTC.MS_HOUR:
u_values.append(f"(%(project_id_{i})s,to_timestamp(%(first_recorded_{i})s/1000))") u_values.append(f"(%(project_id_{i})s,to_timestamp(%(first_recorded_{i})s/1000))")
params[f"project_id_{i}"] = r["project_id"] params[f"project_id_{i}"] = r["project_id"]
params[f"first_recorded_{i}"] = r["first_recorded"] if r["recorded"] else None params[f"first_recorded_{i}"] = r["first_recorded"] if r["recorded"] else None
r.pop("first_recorded_session_at") r.pop("first_recorded_session_at")
r.pop("first_recorded") r.pop("first_recorded")
r.pop("sessions_last_check_at")
if len(u_values) > 0: if len(u_values) > 0:
query = cur.mogrify(f"""UPDATE public.projects query = cur.mogrify(f"""UPDATE public.projects
SET sessions_last_check_at=(now() at time zone 'utc'), first_recorded_session_at=u.first_recorded SET sessions_last_check_at=(now() at time zone 'utc'), first_recorded_session_at=u.first_recorded
@ -120,26 +114,7 @@ def get_projects(tenant_id, recording_state=False, gdpr=None, recorded=False, st
else: else:
for r in rows: for r in rows:
r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"]) r["created_at"] = TimeUTC.datetime_to_timestamp(r["created_at"])
if recording_state and len(rows) > 0: r.pop("sessions_last_check_at")
project_ids = [f'({r["project_id"]})' for r in rows]
query = cur.mogrify(f"""SELECT projects.project_id, COALESCE(MAX(start_ts), 0) AS last
FROM (VALUES {",".join(project_ids)}) AS projects(project_id)
LEFT JOIN sessions USING (project_id)
WHERE sessions.start_ts >= %(startDate)s AND sessions.start_ts <= %(endDate)s
GROUP BY project_id;""",
{"startDate": TimeUTC.now(delta_days=-3), "endDate": TimeUTC.now(delta_days=1)})
cur.execute(query=query)
status = cur.fetchall()
for r in rows:
r["status"] = "red"
for s in status:
if s["project_id"] == r["project_id"]:
if TimeUTC.now(-2) <= s["last"] < TimeUTC.now(-1):
r["status"] = "yellow"
elif s["last"] >= TimeUTC.now(-1):
r["status"] = "green"
break
return helper.list_to_camel_case(rows) return helper.list_to_camel_case(rows)

2
ee/api/chalicelib/core/sessions_exp.py
View file

@ -1272,7 +1272,7 @@ def search_query_parts_ch(data: schemas.SessionsSearchPayloadSchema, error_statu
def search_by_metadata(tenant_id, user_id, m_key, m_value, project_id=None): def search_by_metadata(tenant_id, user_id, m_key, m_value, project_id=None):
if project_id is None: if project_id is None:
all_projects = projects.get_projects(tenant_id=tenant_id, recording_state=False) all_projects = projects.get_projects(tenant_id=tenant_id)
else: else:
all_projects = [ all_projects = [
projects.get_project(tenant_id=tenant_id, project_id=int(project_id), include_last_session=False, projects.get_project(tenant_id=tenant_id, project_id=int(project_id), include_last_session=False,

3
ee/api/chalicelib/core/users.py
View file

@ -575,8 +575,7 @@ def set_password_invitation(tenant_id, user_id, new_password):
c = tenants.get_by_tenant_id(tenant_id) c = tenants.get_by_tenant_id(tenant_id)
c.pop("createdAt") c.pop("createdAt")
c["projects"] = projects.get_projects(tenant_id=tenant_id, recording_state=True, recorded=True, c["projects"] = projects.get_projects(tenant_id=tenant_id, recorded=True, user_id=user_id)
stack_integrations=True, user_id=user_id)
c["smtp"] = helper.has_smtp() c["smtp"] = helper.has_smtp()
c["iceServers"] = assist.get_ice_servers() c["iceServers"] = assist.get_ice_servers()
return { return {

4
ee/api/routers/core_dynamic.py
View file

@ -159,8 +159,8 @@ async def search_sessions_by_metadata(key: str, value: str, projectId: Optional[
@app.get('/projects', tags=['projects']) @app.get('/projects', tags=['projects'])
async def get_projects(context: schemas.CurrentContext = Depends(OR_context)): async def get_projects(context: schemas.CurrentContext = Depends(OR_context)):
return {"data": projects.get_projects(tenant_id=context.tenant_id, recording_state=True, gdpr=True, recorded=True, return {"data": projects.get_projects(tenant_id=context.tenant_id, gdpr=True,
stack_integrations=True, user_id=context.user_id)} recorded=True, user_id=context.user_id)}
# for backward compatibility # for backward compatibility