Coverage for changes_metadata_manager / zenodo_api.py: 100%
79 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-07-17 16:43 +0000
1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import json
6import sqlite3
7import time
8from pathlib import Path
10import requests
11from piccione.upload.on_zenodo import get_headers
12from rich.console import Console
14from changes_metadata_manager.zenodo_metadata import ZenodoUpdatePayload
16console = Console()
18MAX_RETRIES = 5
19BASE_BACKOFF = 10
20REQUEST_TIMEOUT = 120
21RETRYABLE_STATUS_CODES = {429, 502, 503, 504}
22CACHE_TTL_SECONDS = 24 * 60 * 60
25class ZenodoRecordCache:
26 def __init__(self, path: Path, ttl_seconds: int = CACHE_TTL_SECONDS) -> None:
27 self.ttl_seconds = ttl_seconds
28 self.connection = sqlite3.connect(path)
29 self.connection.execute(
30 """
31 CREATE TABLE IF NOT EXISTS records (
32 zenodo_url TEXT NOT NULL,
33 record_id TEXT NOT NULL,
34 fetched_at REAL NOT NULL,
35 has_edit_draft INTEGER NOT NULL,
36 record_json TEXT NOT NULL,
37 PRIMARY KEY (zenodo_url, record_id)
38 )
39 """
40 )
41 self.connection.commit()
43 def __enter__(self) -> "ZenodoRecordCache":
44 return self
46 def __exit__(
47 self,
48 exc_type: type[BaseException] | None,
49 exc_value: BaseException | None,
50 traceback: object | None,
51 ) -> None:
52 self.connection.close()
54 def get(self, zenodo_url: str, record_id: str) -> tuple[dict, bool] | None:
55 normalized_url = zenodo_url.rstrip("/")
56 row = self.connection.execute(
57 """
58 SELECT fetched_at, has_edit_draft, record_json
59 FROM records
60 WHERE zenodo_url = ? AND record_id = ?
61 """,
62 (normalized_url, record_id),
63 ).fetchone()
64 if row is None:
65 return None
67 fetched_at, has_edit_draft, record_json = row
68 if time.time() - fetched_at >= self.ttl_seconds:
69 self.invalidate(normalized_url, record_id)
70 return None
71 return json.loads(record_json), bool(has_edit_draft)
73 def set(
74 self,
75 zenodo_url: str,
76 record_id: str,
77 record: dict,
78 has_edit_draft: bool,
79 ) -> None:
80 self.connection.execute(
81 """
82 INSERT INTO records (
83 zenodo_url, record_id, fetched_at, has_edit_draft, record_json
84 ) VALUES (?, ?, ?, ?, ?)
85 ON CONFLICT (zenodo_url, record_id) DO UPDATE SET
86 fetched_at = excluded.fetched_at,
87 has_edit_draft = excluded.has_edit_draft,
88 record_json = excluded.record_json
89 """,
90 (
91 zenodo_url.rstrip("/"),
92 record_id,
93 time.time(),
94 has_edit_draft,
95 json.dumps(record),
96 ),
97 )
98 self.connection.commit()
100 def invalidate(self, zenodo_url: str, record_id: str) -> None:
101 self.connection.execute(
102 "DELETE FROM records WHERE zenodo_url = ? AND record_id = ?",
103 (zenodo_url.rstrip("/"), record_id),
104 )
105 self.connection.commit()
108def request_with_retry(method: str, url: str, **kwargs) -> requests.Response:
109 response = requests.request(method, url, **kwargs)
110 for attempt in range(1, MAX_RETRIES):
111 if response.status_code not in RETRYABLE_STATUS_CODES:
112 return response
113 wait = BASE_BACKOFF * (2**attempt)
114 console.print(
115 f" [yellow]HTTP {response.status_code}, retrying in {wait}s...[/yellow]"
116 )
117 time.sleep(wait)
118 response = requests.request(method, url, **kwargs)
119 return response
122def fetch_record(
123 zenodo_url: str, record_id: str, access_token: str, user_agent: str
124) -> tuple[dict, bool]:
125 headers = get_headers(access_token, user_agent)
126 headers["Accept"] = "application/vnd.inveniordm.v1+json"
127 response = request_with_retry(
128 "GET",
129 f"{zenodo_url}/records/{record_id}/draft",
130 headers=headers,
131 timeout=REQUEST_TIMEOUT,
132 )
133 has_edit_draft = response.status_code != 404
134 if not has_edit_draft:
135 return (
136 fetch_published_record(
137 zenodo_url,
138 record_id,
139 access_token,
140 user_agent,
141 ),
142 False,
143 )
144 response.raise_for_status()
145 return response.json(), True
148def fetch_published_record(
149 zenodo_url: str, record_id: str, access_token: str, user_agent: str
150) -> dict:
151 return _fetch_published_record(
152 f"{zenodo_url}/records/{record_id}",
153 access_token,
154 user_agent,
155 )
158def fetch_latest_published_record(
159 zenodo_url: str, record_id: str, access_token: str, user_agent: str
160) -> dict:
161 return _fetch_published_record(
162 f"{zenodo_url}/records/{record_id}/versions/latest",
163 access_token,
164 user_agent,
165 )
168def _fetch_published_record(
169 record_url: str, access_token: str, user_agent: str
170) -> dict:
171 headers = get_headers(access_token, user_agent)
172 headers["Accept"] = "application/vnd.inveniordm.v1+json"
173 response = request_with_retry(
174 "GET",
175 record_url,
176 headers=headers,
177 timeout=REQUEST_TIMEOUT,
178 )
179 response.raise_for_status()
180 return response.json()
183def create_edit_draft(
184 zenodo_url: str, record_id: str, access_token: str, user_agent: str
185) -> dict:
186 response = request_with_retry(
187 "POST",
188 f"{zenodo_url}/records/{record_id}/draft",
189 headers=get_headers(access_token, user_agent),
190 timeout=REQUEST_TIMEOUT,
191 )
192 response.raise_for_status()
193 return response.json()
196def update_draft(
197 zenodo_url: str,
198 record_id: str,
199 access_token: str,
200 user_agent: str,
201 payload: ZenodoUpdatePayload,
202) -> None:
203 response = request_with_retry(
204 "PUT",
205 f"{zenodo_url}/records/{record_id}/draft",
206 headers=get_headers(access_token, user_agent, "application/json"),
207 json=payload,
208 timeout=REQUEST_TIMEOUT,
209 )
210 response.raise_for_status()
213def publish_draft(
214 zenodo_url: str, record_id: str, access_token: str, user_agent: str
215) -> None:
216 response = request_with_retry(
217 "POST",
218 f"{zenodo_url}/records/{record_id}/draft/actions/publish",
219 headers=get_headers(access_token, user_agent),
220 timeout=REQUEST_TIMEOUT,
221 )
222 response.raise_for_status()