Coverage for seedboxsync/front/apis/uploads.py: 97%
63 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-11 07:37 +0000
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."""
9from typing import Any
10from flask_restx import Namespace, fields, reqparse
11from peewee import fn
12from seedboxsync.core.database.models import Torrent
13from seedboxsync.front.apis import Resource, parser_period
14from seedboxsync.front.login_manager import login_required
16api = Namespace("uploads", description="Operations related to uploaded torrents management")
19# ==========================
20# Models
21# ==========================
22upload_model = api.model(
23 "Upload",
24 {
25 "id": fields.Integer(required=True, description="Unique identifier of the uploaded torrent", example=99),
26 "name": fields.String(required=True, description="Torrent file name", example="Justo.torrent"),
27 "announce": fields.String(
28 required=False,
29 description="Announce URL or tracker information from the torrent file",
30 example="https://serversecret.com/anounce",
31 ),
32 "announcer": fields.String(
33 required=False,
34 description="Tracker announce domain of the torrent",
35 example="serversecret.com",
36 ),
37 "source": fields.String(
38 required=False,
39 description="Source or provenance of the torrent file",
40 example="serversecret",
41 ),
42 "files": fields.Integer(
43 required=False,
44 description="Total number of files contained in the torrent",
45 example=2,
46 ),
47 "size": fields.Integer(
48 required=False,
49 description="Total size of all files in bytes",
50 example=3337353289,
51 ),
52 "human_size": fields.String(
53 required=False,
54 description="Total size of all files in with related humanization",
55 example="3.1 GiB",
56 ),
57 "private": fields.Boolean(
58 required=False,
59 description="Flag indicating if the torrent is private",
60 example=True,
61 ),
62 "sent": fields.DateTime(
63 dt_format="iso8601",
64 required=True,
65 description="Timestamp when the torrent was uploaded",
66 ),
67 },
68)
69upload_list_envelope = Resource.build_envelope_model(api, "UploadList", nested_model=upload_model)
70upload_envelope = Resource.build_envelope_model(api, "Upload", nested_model=upload_model, as_list=False)
71upload_message_envelope = Resource.build_envelope_model(api, "UploadMessage", as_message=True)
74# ==========================
75# Request parser
76# ==========================
77parser = reqparse.RequestParser()
78parser.add_argument(
79 "offset",
80 type=int,
81 default=0,
82 location="args",
83 help="Number of items to skip before starting to collect the result set (default: 0)",
84)
85parser.add_argument(
86 "limit",
87 type=int,
88 default=50,
89 location="args",
90 help="Maximum number of items to return (min=5, max=1000)",
91)
92parser.add_argument("search", type=str, required=False, help="Optional search string to filter items")
93parser.args.extend(parser_period.args)
96# ==========================
97# Endpoints
98# ==========================
99@api.route("")
100class UploadsList(Resource):
101 """
102 Endpoint to manage uploaded torrents.
104 Provides a list of uploaded torrents with optional limit on the number of items returned.
105 """
107 @api.doc("list_uploads") # type: ignore[untyped-decorator]
108 @api.expect(parser) # type: ignore[untyped-decorator]
109 @api.marshal_with(upload_list_envelope, code=200, description="List of uploaded torrents") # type: ignore[untyped-decorator]
110 @login_required # type: ignore[untyped-decorator]
111 def get(self) -> dict[str, Any]:
112 """
113 Retrieve the most recent uploaded torrents.
115 Query Parameters:
116 - offset: Number of items to skip before starting to collect the result set (default: 0)
117 - limit: Maximum number of downloads to return (default=50)
118 - search: Optional search string to filter items
119 """
120 args = parser.parse_args()
121 offset = args.get("offset")
122 limit = self.set_limit(args.get("limit", 50))
123 search = args.get("search")
124 start_date = args.get("start_date")
125 end_date = args.get("end_date")
127 count = Torrent.select()
128 select = (
129 Torrent.select(
130 Torrent.id,
131 Torrent.name,
132 Torrent.announce,
133 Torrent.announcer,
134 Torrent.source,
135 fn.coalesce(Torrent.total_files, None).alias("files"),
136 fn.coalesce(Torrent.total_size, None).alias("size"),
137 fn.humanize(Torrent.total_size).alias("human_size"),
138 Torrent.private,
139 Torrent.sent,
140 )
141 .limit(limit)
142 .offset(offset)
143 .order_by(Torrent.sent.desc())
144 )
146 if search:
147 count = count.where(Torrent.name.contains(search))
148 select = select.where(Torrent.name.contains(search))
150 if start_date:
151 count = count.where(Torrent.sent >= start_date)
152 select = select.where(Torrent.sent >= start_date)
154 if end_date:
155 count = count.where(Torrent.sent <= end_date)
156 select = select.where(Torrent.sent <= end_date)
158 return self.build_envelope(list(select.dicts()), data_total=count.count(), type="UploadList")
161@api.route("/<int:id>")
162@api.response(404, "Upload not found")
163@api.param("id", "The upload identifier")
164class Uploads(Resource):
165 """
166 Endpoint for managing upload.
168 Provides upload operations.
169 """
171 @api.doc("get_upload") # type: ignore[untyped-decorator]
172 @api.marshal_with(upload_envelope, skip_none=True, code=200, description="Upload element") # type: ignore[untyped-decorator]
173 @login_required # type: ignore[untyped-decorator]
174 def get(self, id: int) -> dict[str, Any]: # noqa: A002
175 """
176 Retrieve an uploaded torrent.
178 Args:
179 id (int): Uploaded torrent identifier.
181 Returns:
182 dict[str, Any]: API response envelope containing the upload.
183 """
184 select: Torrent | None = None
185 try:
186 select = (
187 Torrent.select(
188 Torrent.id,
189 Torrent.name,
190 Torrent.announce,
191 Torrent.announcer,
192 Torrent.source,
193 fn.coalesce(Torrent.total_files, None).alias("files"),
194 fn.coalesce(Torrent.total_size, None).alias("size"),
195 fn.humanize(Torrent.total_size).alias("human_size"),
196 Torrent.private,
197 Torrent.sent,
198 )
199 .where(Torrent.id == id)
200 .dicts()
201 .get()
202 )
204 except Torrent.DoesNotExist: # type: ignore[attr-defined]
205 api.abort(404, f"Upload {id} doesn't exist")
207 return self.build_envelope(select, type="Upload")
209 @api.doc("delete_upload") # type: ignore[untyped-decorator]
210 @api.marshal_with(upload_message_envelope, code=200, description="Delete upload element") # type: ignore[untyped-decorator]
211 @login_required # type: ignore[untyped-decorator]
212 def delete(self, id: int) -> dict[str, Any]: # noqa: A002
213 """
214 Delete an uploaded torrent.
216 Args:
217 id (int): Uploaded torrent identifier.
219 Returns:
220 dict[str, Any]: API response envelope containing a status message.
221 """
222 count = Torrent.delete().where(Torrent.id == id).execute()
223 if count == 0:
224 api.abort(404, f"Upload {id} doesn't exist")
226 return self.build_envelope(None, type="Upload", message=f"Upload {id} deleted.")