Coverage for seedboxsync/front/apis/uploads.py: 90%
83 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-26 17:14 +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 datetime import date
10from typing import Any
11from flask_restx import Namespace, fields, inputs, reqparse
12from peewee import fn
13from seedboxsync.core.database.models import Torrent
14from seedboxsync.front.apis import Resource, parser_period
15from seedboxsync.front.cache import cache
16from seedboxsync.front.login_manager import login_required
18api = Namespace("uploads", description="Operations related to uploaded torrents management")
21# ==========================
22# Models
23# ==========================
24upload_model = api.model(
25 "Upload",
26 {
27 "id": fields.Integer(required=True, description="Unique identifier of the uploaded torrent", example=99),
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 "announcer": fields.String(
35 required=False,
36 description="Tracker announce domain of the torrent",
37 example="serversecret.com",
38 ),
39 "source": fields.String(
40 required=False,
41 description="Source or provenance of the torrent file",
42 example="serversecret",
43 ),
44 "files": fields.Integer(
45 required=False,
46 description="Total number of files contained in the torrent",
47 example=2,
48 ),
49 "size": fields.Integer(
50 required=False,
51 description="Total size of all files in bytes",
52 example=3337353289,
53 ),
54 "human_size": fields.String(
55 required=False,
56 description="Total size of all files in with related humanization",
57 example="3.1 GiB",
58 ),
59 "private": fields.Boolean(
60 required=False,
61 description="Flag indicating if the torrent is private",
62 example=True,
63 ),
64 "sent": fields.DateTime(
65 dt_format="iso8601",
66 required=True,
67 description="Timestamp when the torrent was uploaded",
68 ),
69 },
70)
71upload_list_envelope = Resource.build_envelope_model(api, "UploadList", nested_model=upload_model)
72upload_envelope = Resource.build_envelope_model(api, "Upload", nested_model=upload_model, as_list=False)
73upload_message_envelope = Resource.build_envelope_model(api, "UploadMessage", as_message=True)
75stats_source_model = api.model(
76 "StatsSource",
77 {
78 "source": fields.String(
79 required=True,
80 description="Source of the torrent, falback based on announcer",
81 example="torrenter",
82 ),
83 "total": fields.Integer(
84 required=True,
85 description="Number or size of files for this source",
86 example=4989,
87 ),
88 "total_size": fields.Integer(
89 required=True,
90 description="Size of files for this source",
91 example=21678643250867,
92 ),
93 "human_total_size": fields.String(
94 required=True,
95 description="Total size of files with related source",
96 example="19.7 Tio",
97 ),
98 },
99)
100stats_source_envelope = Resource.build_envelope_model(api, "StatsSource", nested_model=stats_source_model)
103# ==========================
104# Request parser
105# ==========================
106parser = reqparse.RequestParser()
107parser.add_argument(
108 "offset",
109 type=int,
110 default=0,
111 location="args",
112 help="Number of items to skip before starting to collect the result set (default: 0)",
113)
114parser.add_argument(
115 "limit",
116 type=int,
117 default=50,
118 location="args",
119 help="Maximum number of items to return (min=5, max=1000)",
120)
121parser.add_argument(
122 "start_date",
123 type=inputs.date_from_iso8601,
124 location="args",
125 help="Start date for filtering in ISO 8601 format (e.g. YYYY-MM-DD)",
126)
127parser.add_argument(
128 "end_date",
129 type=inputs.date_from_iso8601,
130 location="args",
131 help="End date for filtering in ISO 8601 format (e.g. YYYY-MM-DD)",
132)
133parser.add_argument("search", type=str, required=False, help="Optional search string to filter items")
136# ==========================
137# Endpoints
138# ==========================
139@api.route("")
140class UploadsList(Resource):
141 """
142 Endpoint to manage uploaded torrents.
144 Provides a list of uploaded torrents with optional limit on the number of items returned.
145 """
147 @api.doc("list_uploads") # type: ignore[untyped-decorator]
148 @api.expect(parser) # type: ignore[untyped-decorator]
149 @api.marshal_with(upload_list_envelope, code=200, description="List of uploaded torrents") # type: ignore[untyped-decorator]
150 @login_required # type: ignore[untyped-decorator]
151 def get(self) -> dict[str, Any]:
152 """
153 Retrieve the most recent uploaded torrents.
155 Query Parameters:
156 - offset: Number of items to skip before starting to collect the result set (default: 0)
157 - limit: Maximum number of downloads to return (default=50)
158 - search: Optional search string to filter items
159 """
160 args = parser.parse_args()
161 offset = args.get("offset")
162 limit = self.set_limit(args.get("limit", 50))
163 search = args.get("search")
164 start_date = args.get("start_date")
165 end_date = args.get("end_date")
167 count = Torrent.select()
168 select = (
169 Torrent.select(
170 Torrent.id,
171 Torrent.name,
172 Torrent.announce,
173 Torrent.announcer,
174 Torrent.source,
175 fn.coalesce(Torrent.total_files, None).alias("files"),
176 fn.coalesce(Torrent.total_size, None).alias("size"),
177 fn.humanize(Torrent.total_size).alias("human_size"),
178 Torrent.private,
179 Torrent.sent,
180 )
181 .limit(limit)
182 .offset(offset)
183 .order_by(Torrent.sent.desc())
184 )
186 if search:
187 count = count.where(Torrent.name.contains(search))
188 select = select.where(Torrent.name.contains(search))
190 if start_date:
191 count = count.where(Torrent.sent >= start_date)
192 select = select.where(Torrent.sent >= start_date)
194 if end_date:
195 count = count.where(Torrent.sent <= end_date)
196 select = select.where(Torrent.sent <= end_date)
198 return self.build_envelope(list(select.dicts()), data_total=count.count(), type="Upload")
201@api.route("/<int:id>")
202@api.response(404, "Upload not found")
203@api.param("id", "The upload identifier")
204class Uploads(Resource):
205 """
206 Endpoint for managing upload.
208 Provides upload operations.
209 """
211 @api.doc("get_upload") # type: ignore[untyped-decorator]
212 @api.marshal_with(upload_envelope, skip_none=True, code=200, description="Upload element") # type: ignore[untyped-decorator]
213 @login_required # type: ignore[untyped-decorator]
214 def get(self, id: int) -> dict[str, Any]: # noqa: A002
215 """
216 Retrieve an uploaded torrent.
218 Args:
219 id (int): Uploaded torrent identifier.
221 Returns:
222 dict[str, Any]: API response envelope containing the upload.
223 """
224 select: Torrent | None = None
225 try:
226 select = (
227 Torrent.select(
228 Torrent.id,
229 Torrent.name,
230 Torrent.announce,
231 Torrent.announcer,
232 Torrent.source,
233 fn.coalesce(Torrent.total_files, None).alias("files"),
234 fn.coalesce(Torrent.total_size, None).alias("size"),
235 fn.humanize(Torrent.total_size).alias("human_size"),
236 Torrent.private,
237 Torrent.sent,
238 )
239 .where(Torrent.id == id)
240 .dicts()
241 .get()
242 )
244 except Torrent.DoesNotExist: # type: ignore[attr-defined]
245 api.abort(404, f"Upload {id} doesn't exist")
247 return self.build_envelope(select, type="Upload")
249 @api.doc("delete_upload") # type: ignore[untyped-decorator]
250 @api.marshal_with(upload_message_envelope, code=200, description="Delete upload element") # type: ignore[untyped-decorator]
251 @login_required # type: ignore[untyped-decorator]
252 def delete(self, id: int) -> dict[str, Any]: # noqa: A002
253 """
254 Delete an uploaded torrent.
256 Args:
257 id (int): Uploaded torrent identifier.
259 Returns:
260 dict[str, Any]: API response envelope containing a status message.
261 """
262 count = Torrent.delete().where(Torrent.id == id).execute()
263 if count == 0:
264 api.abort(404, f"Upload {id} doesn't exist")
266 return self.build_envelope(None, type="Upload", message=f"Upload {id} deleted.")
269@api.route("/stats/source")
270class UploadsStatsBySource(Resource):
271 """Resource endpoint to retrieve torrent source statistics."""
273 @api.doc("stats_uploads_by_source") # type: ignore[untyped-decorator]
274 @api.marshal_with(stats_source_envelope, code=200, description="Upload statistics aggregated by source") # type: ignore[untyped-decorator]
275 @api.expect(parser_period) # type: ignore[untyped-decorator]
276 @login_required # type: ignore[untyped-decorator]
277 def get(self) -> dict[str, Any]:
278 """
279 Retrieve torrent statistics grouped by source.
281 Fetches aggregated file counts and total sizes per source from the cache
282 or database, then wraps the dataset into a standard API response envelope.
284 Returns:
285 dict[str, Any]: Envelope containing source statistics, metadata,
286 and total element count.
287 """
288 args = parser_period.parse_args()
289 start_date = args.get("start_date")
290 end_date = args.get("end_date")
292 stats = _get_stats_by_source(start_date, end_date)
294 return self.build_envelope(stats, data_total=len(stats), type="StatsSource")
297@cache.memoize(timeout=300)
298def _get_stats_by_source(start_date: date | None, end_date: date | None) -> list[dict[str, object]]:
299 """
300 Fetch torrent statistics grouped by source within an optional date range.
302 Executes the database query to aggregate torrent statistics by source domain
303 filtered by date boundaries if provided, then caches the result using Flask-Caching memoization[cite: 2].
305 Args:
306 start_date (date | None, optional): Optional lower date boundary for filtering. Defaults to None.
307 end_date (date | None, optional): Optional upper date boundary for filtering. Defaults to None.
309 Returns:
310 list[dict[str, object]]: A list of dictionaries containing source statistics,
311 including total counts and associated sizes[cite: 2].
312 """
313 return Torrent.get_stats_by_source(start_date, end_date)