Coverage for datacite/views.py: 99%

161 statements  

« prev     ^ index     » next       coverage.py v7.10.1, created at 2025-07-29 15:38 +0000

1import json 

2import logging 

3from typing import Any 

4 

5import requests 

6from django.conf import settings 

7from django.contrib import messages 

8from django.contrib.messages.views import SuccessMessageMixin 

9from django.db.models import F, IntegerField, Prefetch, Value 

10from django.http import HttpResponse 

11from django.template.loader import render_to_string 

12from django.views.generic import ListView 

13from django.views.generic.detail import DetailView 

14from django.views.generic.edit import CreateView 

15from extra_views import ( # type: ignore[import-untyped] 

16 CreateWithInlinesView, 

17 UpdateWithInlinesView, 

18) 

19from extra_views.contrib.mixins import ( # type: ignore[import-untyped] 

20 SuccessMessageWithInlinesMixin, 

21) 

22 

23from datacite.datacite import DataciteRESTClient 

24from datacite.forms import ( 

25 ContributorInline, 

26 CreatorForm, 

27 DescriptionInline, 

28 FunderForm, 

29 FundingForm, 

30 GeolocationInline, 

31 IdentifierInline, 

32 InlineFormSetHelper, 

33 MetadataForm, 

34 RightsForm, 

35 TitleInline, 

36) 

37from datacite.models import ( 

38 DEFAULT_LANGUAGE, 

39 Funding, 

40 Metadata, 

41 MetadataContributor, 

42 Participant, 

43 ParticipantTypes, 

44 Rights, 

45) 

46from datacite.serializers import MetadataSerializer 

47from network.forms import NetworkForm 

48from network.models import Network 

49 

50FLASH_MESSAGES_HTML = "flash_messages.html" 

51 

52logger = logging.getLogger(__name__) 

53 

54 

55class MetadataList(ListView): 

56 model = Network 

57 template_name = "datacite/metadata_list.html" 

58 

59 def get_queryset(self) -> Any: 

60 return super().get_queryset().order_by("fdsn_code", "start_year") 

61 

62 

63class MetadataDetail(DetailView): 

64 model = Network 

65 template_name = "datacite/metadata_detail.html" 

66 string_error_format_for_datacite = ( 

67 "Failed to upsert metadata for %s in Datacite : %s" 

68 ) 

69 string_error_format_for_unreachable_datacite = ( 

70 f"Failed to publish the metadata in Datacite : {0}" 

71 ) 

72 known_http_error_message = "Connection error with datacite.org" 

73 server_error_message = "Server error from datacite.org" 

74 

75 def post(self, request: Any, *args: Any, **kwargs: Any) -> Any: 

76 """ 

77 Publish the metadata of the current object to Datacite. It returns to the 

78 detail page. 

79 """ 

80 response = super().get(request, *args, **kwargs) 

81 doi = self.object.doi 

82 try: 

83 DataciteRESTClient().upsert_doi( 

84 doi=doi, 

85 doi_metadata=MetadataSerializer(instance=self.object.metadata).data, 

86 ) 

87 

88 logger.info("Successfully upserted metadata for %s in Datacite.", doi) 

89 messages.success( 

90 self.request, "Successfully published the metadata in Datacite." 

91 ) 

92 except ( 

93 requests.exceptions.ConnectionError, 

94 requests.exceptions.Timeout, 

95 requests.exceptions.TooManyRedirects, 

96 ): 

97 message = "Connection error with datacite.org" 

98 logger.warning( 

99 self.string_error_format_for_datacite, 

100 doi, 

101 message, 

102 ) 

103 messages.error( 

104 self.request, 

105 self.string_error_format_for_unreachable_datacite.format( 

106 self.known_http_error_message 

107 ), 

108 ) 

109 except requests.exceptions.HTTPError as exception: 

110 if ( 

111 requests.codes.bad_request 

112 <= exception.response.status_code 

113 < requests.codes.server_error 

114 ): 

115 logger.warning( 

116 self.string_error_format_for_datacite, 

117 doi, 

118 exception.response.json()["errors"], 

119 ) 

120 for error in exception.response.json()["errors"]: 

121 messages.error(self.request, error.__str__()) 

122 else: 

123 logger.warning( 

124 self.string_error_format_for_datacite, 

125 doi, 

126 self.server_error_message, 

127 ) 

128 messages.error( 

129 self.request, 

130 self.string_error_format_for_unreachable_datacite.format( 

131 self.server_error_message 

132 ), 

133 ) 

134 

135 return response 

136 

137 def get_queryset(self) -> Any: 

138 """This views model is Network, we have to fetch all information about the 

139 corresponding metadata""" 

140 return ( 

141 super() 

142 .get_queryset() 

143 .prefetch_related( 

144 Prefetch( 

145 "metadata", 

146 queryset=Metadata.objects.prefetch_related( 

147 Prefetch( 

148 "creators", 

149 queryset=Participant.objects.order_by( 

150 "metadatacreator__order" 

151 ), 

152 ), 

153 Prefetch( 

154 "metadatacontributor_set", 

155 queryset=MetadataContributor.objects.select_related( 

156 "contributor" 

157 ), 

158 ), 

159 "title_set", 

160 "description_set", 

161 "geolocation_set", 

162 "formats", 

163 "rights", 

164 ).select_related("network", "publisher", "types"), 

165 ) 

166 ) 

167 ) 

168 

169 

170class MetadataUpdate(SuccessMessageWithInlinesMixin, UpdateWithInlinesView): 

171 model = Metadata 

172 form_class = MetadataForm 

173 inlines = [TitleInline, DescriptionInline, ContributorInline, GeolocationInline] 

174 template_name = "datacite/metadata_form.html" 

175 success_message = "Successfully updated the metadata." 

176 

177 def get_queryset(self) -> Any: 

178 return ( 

179 super() 

180 .get_queryset() 

181 .prefetch_related("title_set", "description_set", "geolocation_set") 

182 .select_related("network", "publisher", "types") 

183 ) 

184 

185 def get_context_data(self, **kwargs: Any) -> Any: 

186 context = super().get_context_data(**kwargs) 

187 context["formset_helper"] = InlineFormSetHelper() 

188 # This change of queryset is necessary to override the default order of 

189 # Creators in the combobox, which is id-ordered by default. 

190 context["form"].fields["creators"].queryset = ( 

191 Participant.objects.prefetch_related("metadatacreator_set") 

192 .filter(metadatacreator__metadata=self.object) 

193 .annotate(order=F("metadatacreator__order")) 

194 .union( 

195 Participant.objects.exclude( 

196 metadatacreator__metadata=self.object 

197 ).annotate(order=Value(None, output_field=IntegerField())) 

198 ) 

199 .order_by(F("order").asc(nulls_last=True), "id") 

200 ) 

201 return context 

202 

203 def form_valid(self, form: MetadataForm) -> Any: 

204 form_valid_resp = super().form_valid(form) 

205 form_kwargs = self.get_form_kwargs() 

206 if "data" in form_kwargs: 206 ↛ 209line 206 didn't jump to line 209 because the condition on line 206 was always true

207 self.manage_creators(data=form_kwargs["data"]) 

208 

209 return form_valid_resp 

210 

211 def manage_creators(self, data: Any) -> None: 

212 creator_ids = data.getlist("creators", []) 

213 self.object.add_ordered_creators_from_ids(creator_ids) 

214 

215 

216class CreatorCreate(CreateWithInlinesView): 

217 model = Participant 

218 inlines = [IdentifierInline] 

219 template_name = "datacite/creator_form.html" 

220 form_class = CreatorForm 

221 initial = {"lang": DEFAULT_LANGUAGE, "name_type": ParticipantTypes.ORGANIZATION} 

222 success_url = "/" 

223 

224 def form_valid(self, form: CreatorForm) -> HttpResponse: 

225 super().form_valid(form) 

226 

227 messages.success(self.request, "Successfully added the creator.") 

228 return HttpResponse( 

229 status=204, 

230 headers={ 

231 "HX-Trigger": json.dumps( 

232 { 

233 "newParticipantAdded": { 

234 "value": self.object.pk, 

235 "text": str(self.object), 

236 "type": self.request.POST.get("type", "creator"), 

237 }, 

238 "showMessage": render_to_string( 

239 template_name=FLASH_MESSAGES_HTML, 

240 context=self.get_context_data(), 

241 request=self.request, 

242 ), 

243 } 

244 ) 

245 }, 

246 ) 

247 

248 def get_context_data(self, **kwargs: Any) -> Any: 

249 context = super().get_context_data(**kwargs) 

250 context["creator_form"] = context.get("form") 

251 context["creator_inlines"] = context.get("inlines") 

252 context["formset_helper"] = InlineFormSetHelper() 

253 context["participant_type"] = self.request.GET.get("type", "creator") 

254 return context 

255 

256 

257class RightsCreate(CreateWithInlinesView): 

258 model = Rights 

259 inlines = [IdentifierInline] 

260 template_name = "datacite/rights_form.html" 

261 form_class = RightsForm 

262 initial = {"lang": DEFAULT_LANGUAGE} 

263 success_url = "/" 

264 

265 def form_valid(self, form: RightsForm) -> HttpResponse: 

266 super().form_valid(form) 

267 

268 messages.success(self.request, "Right successfully added.") 

269 return HttpResponse( 

270 status=204, 

271 headers={ 

272 "HX-Trigger": json.dumps( 

273 { 

274 "newRightsAdded": { 

275 "value": self.object.pk, 

276 "text": str(self.object), 

277 }, 

278 "showMessage": render_to_string( 

279 template_name=FLASH_MESSAGES_HTML, 

280 context=self.get_context_data(), 

281 request=self.request, 

282 ), 

283 } 

284 ) 

285 }, 

286 ) 

287 

288 def get_context_data(self, **kwargs: Any) -> Any: 

289 context = super().get_context_data(**kwargs) 

290 context["rights_form"] = context.get("form") 

291 context["rights_inlines"] = context.get("inlines") 

292 context["formset_helper"] = InlineFormSetHelper() 

293 return context 

294 

295 

296class FunderCreate(CreateWithInlinesView): 

297 model = Participant 

298 template_name = "datacite/funder_form.html" 

299 form_class = FunderForm 

300 initial = {"lang": DEFAULT_LANGUAGE} 

301 inlines = [IdentifierInline] 

302 success_url = "/" 

303 

304 def form_valid(self, form: FundingForm) -> HttpResponse: 

305 super().form_valid(form) 

306 

307 messages.success(self.request, "Funder successfully added.") 

308 return HttpResponse( 

309 status=204, 

310 headers={ 

311 "HX-Trigger": json.dumps( 

312 { 

313 "newFunderAdded": { 

314 "value": self.object.pk, 

315 "text": str(self.object), 

316 }, 

317 "showMessage": render_to_string( 

318 template_name=FLASH_MESSAGES_HTML, 

319 context=self.get_context_data(), 

320 request=self.request, 

321 ), 

322 } 

323 ) 

324 }, 

325 ) 

326 

327 def get_context_data(self, **kwargs: Any) -> Any: 

328 context = super().get_context_data(**kwargs) 

329 context["funder_form"] = context.get("form") 

330 context["funder_inlines"] = context.get("inlines") 

331 context["formset_helper"] = InlineFormSetHelper() 

332 return context 

333 

334 

335class FundingCreate(CreateWithInlinesView): 

336 model = Funding 

337 template_name = "datacite/funding_form.html" 

338 form_class = FundingForm 

339 success_url = "/" 

340 

341 def form_valid(self, form: FundingForm) -> HttpResponse: 

342 super().form_valid(form) 

343 

344 messages.success(self.request, "Funding successfully added.") 

345 return HttpResponse( 

346 status=204, 

347 headers={ 

348 "HX-Trigger": json.dumps( 

349 { 

350 "newFundingAdded": { 

351 "value": self.object.pk, 

352 "text": str(self.object), 

353 }, 

354 "showMessage": render_to_string( 

355 template_name=FLASH_MESSAGES_HTML, 

356 context=self.get_context_data(), 

357 request=self.request, 

358 ), 

359 } 

360 ) 

361 }, 

362 ) 

363 

364 def get_context_data(self, **kwargs: Any) -> Any: 

365 context = super().get_context_data(**kwargs) 

366 context["funding_form"] = context.get("form") 

367 context["formset_helper"] = InlineFormSetHelper() 

368 return context 

369 

370 

371class NetworkCreate(SuccessMessageMixin, CreateView): 

372 model = Network 

373 template_name = "datacite/network_form.html" 

374 form_class = NetworkForm 

375 success_message = "Successfully created the network." 

376 

377 def get_context_data(self, **kwargs: Any) -> dict[str, Any]: 

378 context = super().get_context_data(**kwargs) 

379 context["fdsn_url"] = settings.FDSN_NETWORK_URL 

380 context["existing_networks"] = list( 

381 Network.objects.values("fdsn_code", "start_year") 

382 ) 

383 return context 

384 

385 def form_valid(self, form: NetworkForm) -> HttpResponse: 

386 response = super().form_valid(form) 

387 

388 if self.object and not self.object.doi: 

389 self.object.generate_doi() 

390 

391 if self.object and self.object.supported: 

392 Metadata.objects.create_from_network(network=self.object) 

393 

394 return response