Fetch specific metadatas or metadatasets from the download_metas_xls view (ec4d9dc7) · Commits · certic / Jama

src/resources/models.py

+36 −5

Original line number	Diff line number	Diff line
		@@ -268,12 +268,30 @@ class Project(models.Model):
		self._cached_root_collection = col
		return self._cached_root_collection

		def metadatas(self, exclude_automatic_metas=True) -> List["Metadata"]:
		def metadatas(
		self,
		exclude_automatic_metas: bool = True,
		metadata_ids: Union[List[int], None] = None,
		metadataset_ids: Union[List[int], None] = None,
		) -> List["Metadata"]:
		"""
		Fetches all metadatas available in the project, excluding OCR and ExifTool.

		This is the way to define available columns in a XLSX export.
		"""
		if metadata_ids is not None or metadataset_ids is not None:
		include_query = models.Q()
		if metadata_ids:
		include_query \|= models.Q(pk__in=metadata_ids)
		if metadataset_ids:
		include_query \|= models.Q(set_id__in=metadataset_ids)
		query = Metadata.objects.filter(project=self)
		if include_query:
		query = query.filter(include_query)
		else:
		query = query.none()
		return list(query.select_related("set").order_by("set__title", "title"))

		metadatas = []
		query = (
		MetadataSet.objects.filter(project=self)
		@@ -991,11 +1009,17 @@ class Collection(models.Model):
		titles.append(self.title)
		return titles

		def yield_resource_data_for_export(self) -> Iterator[dict]:
		def yield_resource_data_for_export(
		self,
		metadata_ids: Union[List[int], None] = None,
		metadataset_ids: Union[List[int], None] = None,
		) -> Iterator[dict]:
		known_metadatas = {}
		known_metadatas_labels = []
		for metadata in self.project.metadatas(
		not settings.JAMA_XLSX_EXPORT_AUTOMATIC_METADATAS
		not settings.JAMA_XLSX_EXPORT_AUTOMATIC_METADATAS,
		metadata_ids=metadata_ids,
		metadataset_ids=metadataset_ids,
		):
		known_metadatas[metadata.pk] = metadata
		known_metadatas_labels.append(f"{str(metadata)}")
		@@ -1008,7 +1032,12 @@ class Collection(models.Model):
		for res in self.descendants_resources():
		yield _flatten_resource(res, known_metadatas, known_metadatas_labels)

		def export_to_xlsx(self, xlsx_path: str):
		def export_to_xlsx(
		self,
		xlsx_path: str,
		metadata_ids: Union[List[int], None] = None,
		metadataset_ids: Union[List[int], None] = None,
		):
		from openpyxl import Workbook
		from openpyxl.utils.exceptions import IllegalCharacterError

		@@ -1016,7 +1045,9 @@ class Collection(models.Model):
		ws = wb.active
		row_number = 2 # /!\ Index starts at 1. Leave the first row for headers.
		r = {}
		for r in self.yield_resource_data_for_export():
		for r in self.yield_resource_data_for_export(
		metadata_ids=metadata_ids, metadataset_ids=metadataset_ids
		):
		col_number = 1
		for value in r.values():
		cell = ws.cell(row=row_number, column=col_number)

src/rpc/tests.py

+68 −0

Original line number	Diff line number	Diff line
		@@ -7,6 +7,9 @@ from rpc.methods import ServiceException
		from resources.acl import UserAccess
		from pathlib import Path
		import os
		from io import BytesIO
		from django.urls import reverse
		from openpyxl import load_workbook

		object_classes = [
		"collection",
		@@ -360,3 +363,68 @@ class ServiceTestCase(TestCase):
		self.assertTrue(
		models.Resource.objects.filter(deleted_at__isnull=True).count() == 1
		)

		def test_download_metas_xls_filters_metadatas_and_metadatasets(self):
		collection = models.Collection.objects.create(
		title="xlsx target",
		project=self.test_project,
		parent=self.test_project.root_collection,
		)
		ds1 = models.MetadataSet.objects.create(
		project=self.test_project, title="set A"
		)
		ds2 = models.MetadataSet.objects.create(
		project=self.test_project, title="set B"
		)
		ds3 = models.MetadataSet.objects.create(
		project=self.test_project, title="set C"
		)
		m1 = models.Metadata.objects.create(
		title="meta A", set=ds1, project=self.test_project
		)
		m2 = models.Metadata.objects.create(
		title="meta B", set=ds2, project=self.test_project
		)
		m3 = models.Metadata.objects.create(
		title="meta C", set=ds3, project=self.test_project
		)
		models.MetadataCollectionValue.objects.create(
		metadata=m1, collection=collection, value="value A"
		)
		models.MetadataCollectionValue.objects.create(
		metadata=m2, collection=collection, value="value B"
		)
		models.MetadataCollectionValue.objects.create(
		metadata=m3, collection=collection, value="value C"
		)

		self.client.force_login(self.test_user)
		response = self.client.get(
		reverse("metas_download", kwargs={"collection_id": collection.pk}),
		{"metadata_ids": str(m1.pk), "metadataset_ids": str(ds2.pk)},
		)
		self.assertEqual(response.status_code, 200)

		wb = load_workbook(filename=BytesIO(b"".join(response.streaming_content)))
		ws = wb.active
		headers = [ws.cell(row=1, column=i).value for i in range(1, ws.max_column + 1)]
		headers_index = {h: i + 1 for i, h in enumerate(headers)}

		self.assertIn(str(m1), headers)
		self.assertIn(str(m2), headers)
		self.assertNotIn(str(m3), headers)
		self.assertEqual(ws.cell(row=2, column=headers_index[str(m1)]).value, "value A")
		self.assertEqual(ws.cell(row=2, column=headers_index[str(m2)]).value, "value B")

		def test_download_metas_xls_bad_query_params(self):
		collection = models.Collection.objects.create(
		title="xlsx bad query target",
		project=self.test_project,
		parent=self.test_project.root_collection,
		)
		self.client.force_login(self.test_user)
		response = self.client.get(
		reverse("metas_download", kwargs={"collection_id": collection.pk}),
		{"metadata_ids": "abc"},
		)
		self.assertEqual(response.status_code, 400)

src/rpc/views.py

+30 −2

Original line number	Diff line number	Diff line
		@@ -19,7 +19,7 @@ from resources.helpers import (
		ConcurrencyError,
		)
		from resources import models
		from typing import Union
		from typing import Union, List
		from jama import settings
		import re
		import os
		@@ -153,6 +153,25 @@ def _get_project_from_request(request: HttpRequest) -> Union[models.Project, Non
		return None


		def _parse_int_list_query_param(
		request: HttpRequest, key: str
		) -> Union[List[int], None]:
		if key not in request.GET:
		return None
		ids = []
		for raw_val in request.GET.getlist(key):
		for part in raw_val.split(","):
		part = part.strip()
		if not part:
		continue
		value = int(part)
		if value < 1:
		raise ValueError
		ids.append(value)
		# deduplicate while preserving order
		return list(dict.fromkeys(ids))


		@lru_cache(None, typed=True)
		def _get_rpc_methods() -> dict:
		methods = {}
		@@ -479,9 +498,18 @@ def download_metas_xls(request: HttpRequest, collection_id):
		if not user:
		return HttpResponse("Forbidden", status=403)
		UserAccess(user, collection_instance.project).check_read(collection_instance)
		try:
		metadata_ids = _parse_int_list_query_param(request, "metadata_ids")
		metadataset_ids = _parse_int_list_query_param(request, "metadataset_ids")
		except ValueError:
		return HttpResponse("Bad Request", status=400)

		with tempfile.NamedTemporaryFile(suffix=".xlsx") as tmp:
		collection_instance.export_to_xlsx(tmp.name)
		collection_instance.export_to_xlsx(
		tmp.name,
		metadata_ids=metadata_ids,
		metadataset_ids=metadataset_ids,
		)
		return RangedFileResponse(
		request,
		open(tmp.name, "rb"),