Coverage for seedboxsync/front/apis/uploads.py: 97%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""SeedboxSync api uploads view.""" 

8 

9from datetime import timedelta 

10from typing import Any 

11from flask_restx import Namespace, fields, reqparse 

12from peewee import fn 

13from seedboxsync.core.database.models import Torrent 

14from seedboxsync.front.apis import Resource, parser_pagination, parser_period 

15from seedboxsync.front.login_manager import admin_required, login_required 

16 

17api = Namespace("uploads", description="Operations related to uploaded torrents management") 

18 

19 

20# ========================== 

21# Models 

22# ========================== 

23upload_model = api.model( 

24 "Upload", 

25 { 

26 "id": fields.Integer(required=True, description="Unique identifier of the uploaded torrent", example=99), 

27 "name": fields.String(required=True, description="Torrent file name", example="Justo.torrent"), 

28 "announce": fields.String( 

29 required=False, 

30 description="Announce URL or tracker information from the torrent file", 

31 example="https://serversecret.com/anounce", 

32 ), 

33 "announcer": fields.String( 

34 required=False, 

35 description="Tracker announce domain of the torrent", 

36 example="serversecret.com", 

37 ), 

38 "source": fields.String( 

39 required=False, 

40 description="Source or provenance of the torrent file", 

41 example="serversecret", 

42 ), 

43 "files": fields.Integer( 

44 required=False, 

45 description="Total number of files contained in the torrent", 

46 example=2, 

47 ), 

48 "size": fields.Integer( 

49 required=False, 

50 description="Total size of all files in bytes", 

51 example=3337353289, 

52 ), 

53 "human_size": fields.String( 

54 required=False, 

55 description="Total size of all files in with related humanization", 

56 example="3.1 GiB", 

57 ), 

58 "private": fields.Boolean( 

59 required=False, 

60 description="Flag indicating if the torrent is private", 

61 example=True, 

62 ), 

63 "sent": fields.DateTime( 

64 dt_format="iso8601", 

65 required=True, 

66 description="Timestamp when the torrent was uploaded", 

67 ), 

68 }, 

69) 

70upload_list_envelope = Resource.build_envelope_model(api, "UploadList", nested_model=upload_model) 

71upload_envelope = Resource.build_envelope_model(api, "Upload", nested_model=upload_model, as_list=False) 

72upload_message_envelope = Resource.build_envelope_model(api, "UploadMessage", as_message=True) 

73 

74 

75# ========================== 

76# Request parser 

77# ========================== 

78parser = reqparse.RequestParser() 

79parser.add_argument("search", type=str, required=False, help="Optional search string to filter items") 

80parser.args.extend(parser_period.args) 

81parser.args.extend(parser_pagination.args) 

82 

83 

84# ========================== 

85# Endpoints 

86# ========================== 

87@api.route("") 

88class UploadsList(Resource): 

89 """ 

90 Endpoint to manage uploaded torrents. 

91 

92 Provides a list of uploaded torrents with optional limit on the number of items returned. 

93 """ 

94 

95 @api.doc("list_uploads") # type: ignore[untyped-decorator] 

96 @api.expect(parser) # type: ignore[untyped-decorator] 

97 @api.marshal_with(upload_list_envelope, code=200, description="List of uploaded torrents") # type: ignore[untyped-decorator] 

98 @login_required # type: ignore[untyped-decorator] 

99 def get(self) -> dict[str, Any]: 

100 """ 

101 Retrieve the most recent uploaded torrents. 

102 

103 Query Parameters: 

104 - offset: Number of items to skip before starting to collect the result set (default: 0) 

105 - limit: Maximum number of downloads to return (default=50) 

106 - search: Optional search string to filter items 

107 """ 

108 args = parser.parse_args() 

109 offset = args.get("offset") 

110 limit = self.set_limit(args.get("limit", 50)) 

111 search = args.get("search") 

112 start_date = args.get("start_date") 

113 end_date = args.get("end_date") 

114 

115 count = Torrent.select() 

116 select = ( 

117 Torrent.select( 

118 Torrent.id, 

119 Torrent.name, 

120 Torrent.announce, 

121 Torrent.announcer, 

122 Torrent.source, 

123 fn.coalesce(Torrent.total_files, None).alias("files"), 

124 fn.coalesce(Torrent.total_size, None).alias("size"), 

125 fn.humanize(Torrent.total_size).alias("human_size"), 

126 Torrent.private, 

127 Torrent.sent, 

128 ) 

129 .limit(limit) 

130 .offset(offset) 

131 .order_by(Torrent.sent.desc()) 

132 ) 

133 

134 if search: 

135 count = count.where(Torrent.name.contains(search)) 

136 select = select.where(Torrent.name.contains(search)) 

137 

138 if start_date: 

139 count = count.where(Torrent.sent >= start_date) 

140 select = select.where(Torrent.sent >= start_date) 

141 

142 if end_date: 

143 end_exclusive = end_date + timedelta(days=1) 

144 count = count.where(Torrent.sent < end_exclusive) 

145 select = select.where(Torrent.sent < end_exclusive) 

146 

147 return self.build_envelope(list(select.dicts()), data_total=count.count(), type="UploadList") 

148 

149 

150@api.route("/<int:id>") 

151@api.response(404, "Upload not found") 

152@api.param("id", "The upload identifier") 

153class Uploads(Resource): 

154 """ 

155 Endpoint for managing upload. 

156 

157 Provides upload operations. 

158 """ 

159 

160 @api.doc("get_upload") # type: ignore[untyped-decorator] 

161 @api.marshal_with(upload_envelope, skip_none=True, code=200, description="Upload element") # type: ignore[untyped-decorator] 

162 @login_required # type: ignore[untyped-decorator] 

163 def get(self, id: int) -> dict[str, Any]: # noqa: A002 

164 """ 

165 Retrieve an uploaded torrent. 

166 

167 Args: 

168 id (int): Uploaded torrent identifier. 

169 

170 Returns: 

171 dict[str, Any]: API response envelope containing the upload. 

172 """ 

173 select: Torrent | None = None 

174 try: 

175 select = ( 

176 Torrent.select( 

177 Torrent.id, 

178 Torrent.name, 

179 Torrent.announce, 

180 Torrent.announcer, 

181 Torrent.source, 

182 fn.coalesce(Torrent.total_files, None).alias("files"), 

183 fn.coalesce(Torrent.total_size, None).alias("size"), 

184 fn.humanize(Torrent.total_size).alias("human_size"), 

185 Torrent.private, 

186 Torrent.sent, 

187 ) 

188 .where(Torrent.id == id) 

189 .dicts() 

190 .get() 

191 ) 

192 

193 except Torrent.DoesNotExist: # type: ignore[attr-defined] 

194 api.abort(404, f"Upload {id} doesn't exist") 

195 

196 return self.build_envelope(select, type="Upload") 

197 

198 @api.doc("delete_upload") # type: ignore[untyped-decorator] 

199 @api.marshal_with(upload_message_envelope, code=200, description="Delete upload element") # type: ignore[untyped-decorator] 

200 @admin_required 

201 def delete(self, id: int) -> dict[str, Any]: # noqa: A002 

202 """ 

203 Delete an uploaded torrent. 

204 

205 Args: 

206 id (int): Uploaded torrent identifier. 

207 

208 Returns: 

209 dict[str, Any]: API response envelope containing a status message. 

210 """ 

211 count = Torrent.delete().where(Torrent.id == id).execute() 

212 if count == 0: 

213 api.abort(404, f"Upload {id} doesn't exist") 

214 

215 return self.build_envelope(None, type="Upload", message=f"Upload {id} deleted.")