Coverage for seedboxsync/front/apis/downloads.py: 95%
79 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 error module."""
9from typing import Any
10from flask_restx import Namespace, fields, inputs, reqparse
11from peewee import fn
12from seedboxsync.core.database.models import Download
13from seedboxsync.front.apis import DateTimeOrZero, Resource, parser_period
14from seedboxsync.front.login_manager import login_required
16api = Namespace("downloads", description="Operations related to download management")
19# ==========================
20# Models
21# ==========================
22download_model = api.model(
23 "Download",
24 {
25 "id": fields.Integer(
26 required=True,
27 description="Unique identifier of the download record",
28 example=999,
29 ),
30 "path": fields.String(
31 required=True,
32 description="Local path of the downloaded file",
33 example="ConvallisMorbi.doc",
34 ),
35 "mime_extension": fields.String(
36 required=True,
37 description="File extension detected during MIME analysis",
38 example="doc",
39 ),
40 "mime_type": fields.String(
41 required=True,
42 description="MIME type detected for the file (e.g. application/msword)",
43 example="ConvallisMorbi.doc",
44 ),
45 "mime_confidence": fields.String(
46 required=True,
47 description="Confidence level or method used for MIME detection (e.g. extension, magic)",
48 example="puremagic (confidence: 1)",
49 ),
50 "started": fields.DateTime(dt_format="iso8601", required=True, description="Download start timestamp"),
51 "finished": DateTimeOrZero(
52 dt_format="iso8601",
53 required=False,
54 description="Download completion timestamp",
55 ),
56 "local_size": fields.Integer(
57 required=True,
58 description="File size on local storage in bytes",
59 example=3337353289,
60 ),
61 "human_local_size": fields.String(
62 required=True,
63 description="File size on local storage with related humanization",
64 example="3.1 GiB",
65 ),
66 "seedbox_size": fields.Integer(
67 required=True,
68 description="File size on seedbox storage in bytes",
69 example=3337353289,
70 ),
71 "human_seedbox_size": fields.String(
72 required=True,
73 description="File size on seedbox storage with related humanization",
74 example="3.1 GiB",
75 ),
76 "progress": fields.Float(required=True, description="Download progress percentage", example=15.0),
77 },
78)
79download_list_envelope = Resource.build_envelope_model(api, "DownloadList", nested_model=download_model)
80download_envelope = Resource.build_envelope_model(api, "Download", nested_model=download_model, as_list=False)
81download_message_envelope = Resource.build_envelope_model(api, "DownloadMessage", as_message=True)
84# ==========================
85# Request parser
86# ==========================
87parser = reqparse.RequestParser()
88parser.add_argument(
89 "offset",
90 type=int,
91 default=0,
92 location="args",
93 help="Number of items to skip before starting to collect the result set (default: 0)",
94)
95parser.add_argument(
96 "limit",
97 type=int,
98 default=50,
99 location="args",
100 help="Maximum number of items to return (min=5, max=1000)",
101)
102parser.add_argument(
103 "finished",
104 type=inputs.boolean,
105 default=None,
106 location="args",
107 help="Filter only completed downloads (true) or in-progress downloads (false)",
108)
109parser.add_argument("search", type=str, required=False, help="Optional search string to filter items")
110parser.args.extend(parser_period.args)
113# ==========================
114# Endpoints
115# ==========================
116@api.route("")
117class DownloadsList(Resource):
118 """
119 Endpoint for managing downloads list.
121 Provides a list of downloads with optional filtering for in-progress or completed files.
122 """
124 @api.doc("list_downloads") # type: ignore[untyped-decorator]
125 @api.expect(parser) # type: ignore[untyped-decorator]
126 @api.marshal_with(download_list_envelope, code=200, description="List of downloads") # type: ignore[untyped-decorator]
127 @login_required # type: ignore[untyped-decorator]
128 def get(self) -> dict[str, Any]:
129 """
130 Retrieve a list of recent downloads.
132 Query Parameters:
133 - offset: Number of items to skip before starting to collect the result set (default: 0)
134 - limit: Maximum number of downloads to return (default=50)
135 - search: Optional search string to filter items
136 - finished: Filter downloads by status (false=in-progress, true=finished)
137 """
138 args = parser.parse_args()
139 offset = args.get("offset")
140 limit = self.set_limit(args.get("limit", 50))
141 search = args.get("search")
142 finished = args.get("finished")
143 start_date = args.get("start_date")
144 end_date = args.get("end_date")
146 count = Download.select()
147 select = (
148 Download.select(
149 Download.id,
150 Download.path,
151 Download.mime_extension,
152 Download.mime_type,
153 Download.mime_confidence,
154 Download.started,
155 Download.finished,
156 Download.local_size,
157 Download.seedbox_size,
158 fn.humanize(Download.local_size).alias("human_local_size"),
159 fn.humanize(Download.seedbox_size).alias("human_seedbox_size"),
160 fn.round(
161 (Download.local_size.cast("REAL") / Download.seedbox_size.cast("REAL")) * 100,
162 2,
163 ).alias("progress"),
164 )
165 .limit(limit)
166 .offset(offset)
167 .order_by(Download.finished.desc())
168 )
170 if search:
171 count = count.where(Download.path.contains(search))
172 select = select.where(Download.path.contains(search))
174 if finished is not None:
175 # Filter downloads by completion status
176 if finished:
177 count = count.where(Download.finished != 0)
178 select = select.where(Download.finished != 0)
179 else:
180 count = count.where(Download.finished == 0)
181 select = select.where(Download.finished == 0)
183 if start_date:
184 count = count.where(Download.finished >= start_date)
185 select = select.where(Download.finished >= start_date)
187 if end_date:
188 count = count.where(Download.finished <= end_date)
189 select = select.where(Download.finished <= end_date)
191 return self.build_envelope(list(select.dicts()), data_total=count.count(), type="DownloadList")
194@api.route("/progress")
195class DownloadsProgress(Resource):
196 """Endpoint for managing downloads progress."""
198 @api.doc("delete_downloads_progress") # type: ignore[untyped-decorator]
199 @api.marshal_with(download_message_envelope, code=200, description="Downloads in progress deleted") # type: ignore[untyped-decorator]
200 @login_required # type: ignore[untyped-decorator]
201 def delete(self) -> dict[str, Any]:
202 """Delete progress downloads."""
203 count = Download.delete().where(Download.finished == 0).execute()
204 return self.build_envelope(None, type="Download", message=f"{count} download(s) deleted.")
207@api.route("/<int:id>")
208@api.response(404, "Download not found")
209@api.param("id", "The download identifier")
210class Downloads(Resource):
211 """
212 Endpoint for managing downloads.
214 Provides downloads operations.
215 """
217 @api.doc("get_download") # type: ignore[untyped-decorator]
218 @api.marshal_with(download_envelope, skip_none=True, code=200, description="Download element") # type: ignore[untyped-decorator]
219 @login_required # type: ignore[untyped-decorator]
220 def get(self, id: int) -> dict[str, Any]: # noqa: A002
221 """
222 Retrieve a download.
224 Args:
225 id (int): Download identifier.
227 Returns:
228 dict[str, Any]: API response envelope containing the download.
229 """
230 select: Download | None = None
231 try:
232 select = (
233 Download.select(
234 Download.id,
235 Download.path,
236 Download.mime_extension,
237 Download.mime_type,
238 Download.mime_confidence,
239 Download.started,
240 Download.finished,
241 Download.local_size,
242 Download.seedbox_size,
243 fn.humanize(Download.local_size).alias("human_local_size"),
244 fn.humanize(Download.seedbox_size).alias("human_seedbox_size"),
245 fn.round(
246 (Download.local_size.cast("REAL") / Download.seedbox_size.cast("REAL")) * 100,
247 2,
248 ).alias("progress"),
249 )
250 .where(Download.id == id)
251 .dicts()
252 .get()
253 )
254 except Download.DoesNotExist: # type: ignore[attr-defined]
255 api.abort(404, f"Download {id} doesn't exist")
257 return self.build_envelope(select, type="Download")
259 @api.doc("delete_download") # type: ignore[untyped-decorator]
260 @api.marshal_with(download_message_envelope, code=200, description="Delete download element") # type: ignore[untyped-decorator]
261 @login_required # type: ignore[untyped-decorator]
262 def delete(self, id: int) -> dict[str, Any]: # noqa: A002
263 """
264 Delete a download.
266 Args:
267 id (int): Download identifier.
269 Returns:
270 dict[str, Any]: API response envelope containing a status message.
271 """
272 count = Download.delete().where(Download.id == id).execute()
273 if count == 0:
274 api.abort(404, f"Download {id} doesn't exist")
276 return self.build_envelope(None, type="Download", message=f"Download {id} deleted.")