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

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 typing import Any 

10from flask_restx import Namespace, fields, reqparse 

11from seedboxsync.core.dao import Torrent 

12from seedboxsync.front.apis import Resource 

13 

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

15 

16 

17# ========================== 

18# Models 

19# ========================== 

20upload_model = api.model( 

21 "Upload", 

22 { 

23 "id": fields.Integer( 

24 required=True, 

25 description="Unique identifier of the uploaded torrent", 

26 example=99, 

27 ), 

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

29 "announce": fields.String( 

30 required=False, 

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

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

33 ), 

34 "sent": fields.DateTime( 

35 dt_format="iso8601", 

36 required=True, 

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

38 ), 

39 }, 

40) 

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

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

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

44 

45 

46# ========================== 

47# Request parser 

48# ========================== 

49parser = reqparse.RequestParser() 

50parser.add_argument( 

51 "offset", 

52 type=int, 

53 default=0, 

54 location="args", 

55 help="Number of items to skip before starting to collect the result set (default: 0)", 

56) 

57parser.add_argument( 

58 "limit", 

59 type=int, 

60 default=50, 

61 location="args", 

62 help="Maximum number of items to return (min=5, max=1000)", 

63) 

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

65 

66 

67# ========================== 

68# Endpoints 

69# ========================== 

70@api.route("") 

71class UploadsList(Resource): 

72 """ 

73 Endpoint to manage uploaded torrents. 

74 

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

76 """ 

77 

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

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

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

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

82 """ 

83 Retrieve the most recent uploaded torrents. 

84 

85 Query Parameters: 

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

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

88 - search: Optional search string to filter items 

89 """ 

90 args = parser.parse_args() 

91 offset = args.get("offset") 

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

93 search = args.get("search") 

94 

95 count = Torrent.select() 

96 select = Torrent.select(Torrent.id, Torrent.name, Torrent.sent).limit(limit).offset(offset).order_by(Torrent.sent.desc()) 

97 

98 if search: 

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

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

101 

102 return self.build_envelope(list(select.dicts()), data_total=count.count(), type="Upload") 

103 

104 

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

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

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

108class Uploads(Resource): 

109 """ 

110 Endpoint for managing upload. 

111 

112 Provides upload operations. 

113 """ 

114 

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

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

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

118 """ 

119 Retrieve an uploaded torrent. 

120 

121 Args: 

122 id (int): Uploaded torrent identifier. 

123 

124 Returns: 

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

126 """ 

127 select: Torrent | None = None 

128 try: 

129 select = Torrent.select(Torrent.id, Torrent.name, Torrent.sent).where(Torrent.id == id).dicts().get() 

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

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

132 

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

134 

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

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

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

138 """ 

139 Delete an uploaded torrent. 

140 

141 Args: 

142 id (int): Uploaded torrent identifier. 

143 

144 Returns: 

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

146 """ 

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

148 if count == 0: 

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

150 

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