Coverage for cookbook/provider/dropbox.py: 28%

1import io 

2import json 

3import os 

4from datetime import datetime 

5 

6import requests 

7import validators 

8 

9from cookbook.models import Recipe, RecipeImport, SyncLog 

10from cookbook.provider.provider import Provider 

11 

12 

13class Dropbox(Provider): 

14 

15 @staticmethod 

16 def import_all(monitor): 

17 url = "https://api.dropboxapi.com/2/files/list_folder" 

18 

19 headers = { 

20 "Authorization": "Bearer " + monitor.storage.token, 

21 "Content-Type": "application/json" 

22 } 

23 

24 data = { 

25 "path": monitor.path 

26 } 

27 

28 r = requests.post(url, headers=headers, data=json.dumps(data)) 

29 try: 

30 recipes = r.json() 

31 except ValueError: 

32 log_entry = SyncLog(status='ERROR', msg=str(r), sync=monitor) 

33 log_entry.save() 

34 return r 

35 

36 import_count = 0 

37 # TODO check if has_more is set and import that as well 

38 for recipe in recipes['entries']: 

39 path = recipe['path_lower'] 

40 if not Recipe.objects.filter(file_path__iexact=path, space=monitor.space).exists() and not RecipeImport.objects.filter(file_path=path, space=monitor.space).exists(): 

41 name = os.path.splitext(recipe['name'])[0] 

42 new_recipe = RecipeImport( 

43 name=name, 

44 file_path=path, 

45 storage=monitor.storage, 

46 file_uid=recipe['id'], 

47 space=monitor.space, 

48 ) 

49 new_recipe.save() 

50 import_count += 1 

51 

52 log_entry = SyncLog( 

53 status='SUCCESS', 

54 msg='Imported ' + str(import_count) + ' recipes', 

55 sync=monitor, 

56 ) 

57 log_entry.save() 

58 

59 monitor.last_checked = datetime.now() 

60 monitor.save() 

61 

62 return True 

63 

64 @staticmethod 

65 def create_share_link(recipe): 

66 url = "https://api.dropboxapi.com/2/sharing/create_shared_link_with_settings" # noqa: E501 

67 

68 headers = { 

69 "Authorization": "Bearer " + recipe.storage.token, 

70 "Content-Type": "application/json" 

71 } 

72 

73 data = { 

74 "path": recipe.file_uid 

75 } 

76 

77 r = requests.post(url, headers=headers, data=json.dumps(data)) 

78 

79 return r.json() 

80 

81 @staticmethod 

82 def get_share_link(recipe): 

83 url = "https://api.dropboxapi.com/2/sharing/list_shared_links" 

84 

85 headers = { 

86 "Authorization": "Bearer " + recipe.storage.token, 

87 "Content-Type": "application/json" 

88 } 

89 

90 data = { 

91 "path": recipe.file_path, 

92 } 

93 

94 r = requests.post(url, headers=headers, data=json.dumps(data)) 

95 p = r.json() 

96 

97 for link in p['links']: 

98 return link['url'] 

99 

100 response = Dropbox.create_share_link(recipe) 

101 return response['url'] 

102 

103 @staticmethod 

104 def get_file(recipe): 

105 if not recipe.link: 

106 recipe.link = Dropbox.get_share_link(recipe) 

107 recipe.save() 

108 

109 url = recipe.link.replace('www.dropbox.', 'dl.dropboxusercontent.') 

110 if validators.url(url, public=True): 

111 response = requests.get(url) 

112 

113 return io.BytesIO(response.content) 

114 

115 @staticmethod 

116 def rename_file(recipe, new_name): 

117 url = "https://api.dropboxapi.com/2/files/move_v2" 

118 

119 headers = { 

120 "Authorization": "Bearer " + recipe.storage.token, 

121 "Content-Type": "application/json" 

122 } 

123 

124 data = { 

125 "from_path": recipe.file_path, 

126 "to_path": "%s/%s%s" % ( 

127 os.path.dirname(recipe.file_path), 

128 new_name, 

129 os.path.splitext(recipe.file_path)[1] 

130 ) 

131 } 

132 

133 r = requests.post(url, headers=headers, data=json.dumps(data)) 

134 

135 return r.json() 

136 

137 @staticmethod 

138 def delete_file(recipe): 

139 url = "https://api.dropboxapi.com/2/files/delete_v2" 

140 

141 headers = { 

142 "Authorization": "Bearer " + recipe.storage.token, 

143 "Content-Type": "application/json" 

144 } 

145 

146 data = { 

147 "path": recipe.file_path 

148 } 

149 

150 r = requests.post(url, headers=headers, data=json.dumps(data)) 

151 

152 return r.json()