Coverage for chalicelib/core/sessions/sessions_mobs.py: 19%

1from decouple import config 

2 

3from chalicelib.utils.storage import StorageClient 

4 

5 

6def __get_mob_keys(project_id, session_id): 

7 params = { 

8 "sessionId": session_id, 

9 "projectId": project_id 

10 } 

11 return [ 

12 config("SESSION_MOB_PATTERN_S", default="%(sessionId)s") % params, 

13 config("SESSION_MOB_PATTERN_E", default="%(sessionId)se") % params 

14 ] 

15 

16 

17def __get_mobile_video_keys(project_id, session_id): 

18 params = { 

19 "sessionId": session_id, 

20 "projectId": project_id 

21 } 

22 return [ 

23 config("SESSION_IOS_VIDEO_PATTERN") % params, 

24 ] 

25 

26 

27def __get_mob_keys_deprecated(session_id): 

28 return [str(session_id), str(session_id) + "e"] 

29 

30 

31def get_first_url(project_id, session_id, check_existence: bool = True): 

32 k = __get_mob_keys(project_id=project_id, session_id=session_id)[0] 

33 if check_existence and not StorageClient.exists(bucket=config("sessions_bucket"), key=k): 

34 return None 

35 return StorageClient.get_presigned_url_for_sharing( 

36 bucket=config("sessions_bucket"), 

37 expires_in=config("PRESIGNED_URL_EXPIRATION", cast=int, default=900), 

38 key=k 

39 ) 

40 

41 

42def get_urls(project_id, session_id, check_existence: bool = True): 

43 results = [] 

44 for k in __get_mob_keys(project_id=project_id, session_id=session_id): 

45 if check_existence and not StorageClient.exists(bucket=config("sessions_bucket"), key=k): 

46 continue 

47 results.append(StorageClient.get_presigned_url_for_sharing( 

48 bucket=config("sessions_bucket"), 

49 expires_in=config("PRESIGNED_URL_EXPIRATION", cast=int, default=900), 

50 key=k 

51 )) 

52 return results 

53 

54 

55def get_urls_depercated(session_id, check_existence: bool = True): 

56 results = [] 

57 for k in __get_mob_keys_deprecated(session_id=session_id): 

58 if check_existence and not StorageClient.exists(bucket=config("sessions_bucket"), key=k): 

59 continue 

60 results.append(StorageClient.get_presigned_url_for_sharing( 

61 bucket=config("sessions_bucket"), 

62 expires_in=100000, 

63 key=k 

64 )) 

65 return results 

66 

67 

68 

69 

70def delete_mobs(project_id, session_ids): 

71 for session_id in session_ids: 

72 for k in __get_mob_keys(project_id=project_id, session_id=session_id) \ 

73 + __get_mob_keys_deprecated(session_id=session_id): 

74 StorageClient.tag_for_deletion(bucket=config("sessions_bucket"), key=k)