Coverage for src/lanraragi_api/api/database.py: 93%
30 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-22 23:20 +0000
1from os.path import isfile
3from requests import Response
5from lanraragi_api.api.base import (
6 BaseAPICall,
7)
8from lanraragi_api.entity.base import OperationResponse
9from lanraragi_api.entity.database import DatabaseBackup, TagStatistic
10from lanraragi_api.entity.minion import MinionJobResponse
13class DatabaseAPI(BaseAPICall):
14 """Database management APIs.
16 Shared request and error behavior is documented on ``BaseAPICall``.
17 """
19 def get_tag_statistics(
20 self,
21 min_weight: int = 1,
22 hide_excluded_namespaces: bool | None = None,
23 ) -> list[TagStatistic]:
24 """Get tags from the database, with a value symbolizing their prevalence.
26 Args:
27 min_weight: Only get tags whose weight is at least the given
28 minimum. Defaults to 1, which gets all tags.
29 hide_excluded_namespaces: Set to True to exclude tags whose
30 namespace is configured in the server settings. Defaults to
31 None, which returns all tags.
33 Returns:
34 list[TagStatistic]: One entry per tag, with its namespace, text and
35 weight.
37 Raises:
38 APIHttpError: Any status code other than 200.
39 APIResponseDecodeError: If the response body is not valid JSON, or
40 if it is not a list of ``TagStatistic`` objects.
41 """
42 hide_excluded = None
43 if hide_excluded_namespaces is not None:
44 hide_excluded = "true" if hide_excluded_namespaces else "false"
46 return self.request_model_list(
47 "GET",
48 "/api/database/stats",
49 TagStatistic,
50 params={
51 "minweight": min_weight,
52 "hide_excluded_namespaces": hide_excluded,
53 },
54 )
56 def clean_database(self) -> OperationResponse:
57 """Clean the Database.
59 Entries for files that are no longer on the filesystem are hidden and
60 then removed. They are only unlinked at first, so they do not appear in
61 the UI; a subsequent run of this cleanup deletes the unlinked entries.
63 Returns:
64 OperationResponse: Result of the cleanup, including the amount of
65 ``deleted`` and ``unlinked`` entries reported by the server.
67 Raises:
68 APIResponseDecodeError: If the response body is not valid JSON, or
69 does not match ``OperationResponse``.
70 APIOperationError: If the operation failed and raising is enabled.
71 """
72 return self.request_operation("POST", "/api/database/clean")
74 def drop_database(self) -> OperationResponse:
75 """Delete the entire database, including user preferences.
77 This is a rather dangerous endpoint: invoking it might lock you out of
78 the server as a client.
80 Returns:
81 OperationResponse: Result of the database drop.
83 Raises:
84 APIResponseDecodeError: If the response body is not valid JSON, or
85 does not match ``OperationResponse``.
86 APIOperationError: If the operation failed and raising is enabled.
87 """
88 return self.request_operation("POST", "/api/database/drop")
90 def get_backup(self) -> DatabaseBackup:
91 """Scan the entire database and return a backup in JSON form.
93 Consider using ``queue_backup`` if your database is large, as this
94 basic GET endpoint might time out if it takes too long to generate the
95 backup.
97 This backup can be reimported manually through the Backup and Restore
98 feature.
100 Returns:
101 DatabaseBackup: Archive, category and tankoubon metadata of the
102 entire database.
104 Raises:
105 APIHttpError: Any status code other than 200.
106 APIResponseDecodeError: If the response body is not valid JSON, or
107 does not match ``DatabaseBackup``.
108 """
109 return self.request_model("GET", "/api/database/backup", DatabaseBackup)
111 def queue_backup(self) -> MinionJobResponse:
112 """Queue a Minion job to generate a backup JSON file.
114 Use the returned job ID to check progress, then download the file once
115 the job is complete through ``download_backup``.
117 Returns:
118 MinionJobResponse: Enqueued job, whose ``job`` field holds the ID
119 of the Minion job.
121 Raises:
122 APIResponseDecodeError: If the response body is not valid JSON, or
123 does not match ``MinionJobResponse``.
124 APIOperationError: If the operation failed and raising is enabled.
125 """
126 return self.request_operation(
127 "POST", "/api/database/backup", model=MinionJobResponse
128 )
130 def download_backup(self, jobid: int, format: str | None = None) -> Response:
131 """Download the backup JSON file generated by a completed backup job.
133 Args:
134 jobid: ID of the completed backup job.
135 format: Format of the returned backup. ``json`` returns the backup
136 as a JSON response, while ``file`` returns it as a file.
137 Defaults to None, which the server treats as ``file``.
139 Returns:
140 Response: Raw response of the server, holding the backup file or
141 JSON payload.
143 Raises:
144 APIHttpError: 400 if the job is not found or not completed yet, or
145 any other non-2xx status code.
146 """
147 return self.request(
148 "GET",
149 f"/api/database/backup/{jobid}",
150 params={"format": format},
151 )
153 def queue_restore(self, file_path: str) -> MinionJobResponse:
154 """Queue a Minion job to restore from a backup JSON file.
156 Use the returned job ID to check progress.
158 Args:
159 file_path: Path to the backup JSON file to restore. Backslashes are
160 normalized to forward slashes before the file is looked up.
162 Returns:
163 MinionJobResponse: Enqueued job, whose ``job`` field holds the ID
164 of the Minion job.
166 Raises:
167 FileNotFoundError: If ``file_path`` does not point to an existing
168 file.
169 APIResponseDecodeError: If the response body is not valid JSON, or
170 does not match ``MinionJobResponse``.
171 APIOperationError: If the operation failed and raising is enabled.
173 Note:
174 A 400 response, returned for an invalid request such as a malformed
175 backup file, is returned in the operation result, with ``success``
176 set to 0, instead of raising.
177 """
178 file_path = file_path.replace("\\", "/")
180 if not isfile(file_path):
181 raise FileNotFoundError(f"File {file_path} not found")
183 with open(file_path, "rb") as backup_file:
184 return self.request_operation(
185 "POST",
186 "/api/database/restore",
187 model=MinionJobResponse,
188 files={
189 "file": (
190 file_path.split("/")[-1],
191 backup_file,
192 "application/octet-stream",
193 )
194 },
195 )
197 def clear_all_new_flags(self) -> OperationResponse:
198 """Clear the "New!" flag on all archives.
200 Returns:
201 OperationResponse: Result of the flag clearing operation.
203 Raises:
204 APIResponseDecodeError: If the response body is not valid JSON, or
205 does not match ``OperationResponse``.
206 APIOperationError: If the operation failed and raising is enabled.
207 """
208 return self.request_operation("DELETE", "/api/database/isnew")