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