42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
from chalicelib.utils import pg_client, helper
|
|
from chalicelib.utils.storage import StorageClient
|
|
from decouple import config
|
|
|
|
|
|
async def get_test_signals(session_id, project_id):
|
|
async with pg_client.cursor() as cur:
|
|
await cur.execute(cur.mogrify("""\
|
|
SELECT *
|
|
FROM public.ut_tests_signals
|
|
LEFT JOIN public.ut_tests_tasks USING (task_id)
|
|
WHERE session_id = %(session_id)s
|
|
ORDER BY timestamp;""",
|
|
{"project_id": project_id, "session_id": session_id})
|
|
)
|
|
rows = await cur.fetchall()
|
|
return helper.dict_to_camel_case(rows)
|
|
|
|
|
|
async def has_test_signals(session_id, project_id):
|
|
async with pg_client.cursor() as cur:
|
|
await cur.execute(cur.mogrify("""\
|
|
SELECT EXISTS(SELECT 1 FROM public.ut_tests_signals
|
|
WHERE session_id = %(session_id)s) AS has;""",
|
|
{"project_id": project_id, "session_id": session_id})
|
|
)
|
|
row = await cur.fetchone()
|
|
return row.get("has")
|
|
|
|
|
|
async def get_ux_webcam_signed_url(session_id, project_id, check_existence: bool = True):
|
|
results = []
|
|
bucket_name = "uxtesting-records" # config("sessions_bucket")
|
|
k = f'{session_id}/ux_webcam_record.webm'
|
|
if check_existence and not await StorageClient.exists(bucket=bucket_name, key=k):
|
|
return []
|
|
results.append(await StorageClient.get_presigned_url_for_sharing(
|
|
bucket=bucket_name,
|
|
expires_in=100000,
|
|
key=k
|
|
))
|
|
return results
|