From fac694364e44f4efe6423016902d8a604ee7fafa Mon Sep 17 00:00:00 2001 From: Cristiano Singulani Date: Thu, 25 Jun 2026 10:05:36 -0300 Subject: [PATCH 1/4] Add functionality to create and download ZIP files for product directories --- backend/core/test/test_product_download.py | 62 ++++++++++++++++++++++ backend/core/views/product.py | 37 +++++++++---- orchestration | 2 +- 3 files changed, 91 insertions(+), 10 deletions(-) diff --git a/backend/core/test/test_product_download.py b/backend/core/test/test_product_download.py index c2e0448..72d8c4b 100644 --- a/backend/core/test/test_product_download.py +++ b/backend/core/test/test_product_download.py @@ -6,9 +6,11 @@ import yaml from core.models import ( + FileRoles, Product, ProductDownloadArchive, ProductDownloadArchiveStatus, + ProductFile, ProductType, Release, ) @@ -486,6 +488,31 @@ def create_product_source_file(self, media_root): (product_path / "data.csv").write_text("a,b\n1,2\n", encoding="utf-8") return product_path + def create_hats_main_file(self, media_root): + product_path = Path(media_root) / self.product.path + hats_path = product_path / "hats_catalog" + data_path = hats_path / "dataset" / "Norder=0" / "Dir=0" + data_path.mkdir(parents=True) + + (hats_path / "collection.properties").write_text( + "catalog_name=hats_catalog\n", + encoding="utf-8", + ) + (data_path / "Npix=0.parquet").write_bytes(b"parquet") + + ProductFile.objects.create( + product=self.product, + role=FileRoles.MAIN, + name="hats_catalog", + type="application/x-hats", + extension="", + size=0, + file=f"{self.product.path}/hats_catalog", + is_directory=True, + ) + + return hats_path + def test_download_prepare_creates_archive_and_queues_task(self): with tempfile.TemporaryDirectory() as media_root: self.create_product_source_file(media_root) @@ -854,3 +881,38 @@ def test_legacy_download_endpoint_includes_deprecation_headers(self): f"/api/products/{self.product.pk}/download/prepare/", response["Link"], ) + + def test_download_main_file_returns_zip_for_hats_directory(self): + with tempfile.TemporaryDirectory() as media_root: + self.create_hats_main_file(media_root) + + with override_settings(MEDIA_ROOT=media_root): + response = self.client.get( + f"/api/products/{self.product.pk}/download_main_file/" + ) + + content = b"".join(response.streaming_content) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response["Content-Type"], "application/zip") + self.assertIn( + 'filename="hats_catalog.zip"', + response["Content-Disposition"], + ) + + with tempfile.TemporaryDirectory() as tmpdirname: + zip_path = Path(tmpdirname) / "hats_catalog.zip" + zip_path.write_bytes(content) + + with zipfile.ZipFile(zip_path) as archive: + self.assertEqual( + sorted(archive.namelist()), + [ + "collection.properties", + "dataset/Norder=0/Dir=0/Npix=0.parquet", + ], + ) + self.assertEqual( + archive.read("collection.properties").decode("utf-8"), + "catalog_name=hats_catalog\n", + ) diff --git a/backend/core/views/product.py b/backend/core/views/product.py index 78a9c4c..db8bffb 100644 --- a/backend/core/views/product.py +++ b/backend/core/views/product.py @@ -2,6 +2,7 @@ import mimetypes import pathlib import tempfile +import zipfile from json import dumps, loads from pathlib import Path @@ -223,6 +224,32 @@ def __get_full_product(self, product): return product_full + def __build_directory_zip_response(self, directory_path): + zip_handle = tempfile.TemporaryFile() + + with zipfile.ZipFile( + zip_handle, + "w", + compression=zipfile.ZIP_DEFLATED, + compresslevel=ProductDownloadArchiveService.get_compression_level(), + ) as zip_file: + for file_path in sorted(directory_path.rglob("*")): + if not file_path.is_file(): + continue + + arcname = file_path.relative_to(directory_path).as_posix() + zip_file.write(file_path, arcname=arcname) + + size = zip_handle.tell() + zip_handle.seek(0) + + response = FileResponse(zip_handle, content_type="application/zip") + response["Content-Length"] = size + response["Content-Disposition"] = ( + f'attachment; filename="{directory_path.name}.zip"' + ) + return response + def __get_flag_translation_name(self, config): """Extract the flags translation filename from a nested config.""" if isinstance(config, dict): @@ -474,15 +501,7 @@ def download_main_file(self, request, **kwargs): ) if product_path.is_dir(): - return Response( - { - "error": ( - "Main file is a directory-based dataset. " - "Please use the full product download endpoint." - ) - }, - status=status.HTTP_400_BAD_REQUEST, - ) + return self.__build_directory_zip_response(product_path) # Abre o arquivo e envia em bites para o navegador mimetype, _ = mimetypes.guess_type(product_path) diff --git a/orchestration b/orchestration index 48ea57b..145cb70 160000 --- a/orchestration +++ b/orchestration @@ -1 +1 @@ -Subproject commit 48ea57b3f79649be3239bdde7038c3fbe7b320c5 +Subproject commit 145cb70b66af8e1bdcc3e770114aca6d608dce8d From a7099066ab8d96edf02092a7958c556007eca233 Mon Sep 17 00:00:00 2001 From: Cristiano Singulani Date: Thu, 25 Jun 2026 10:26:10 -0300 Subject: [PATCH 2/4] Fix ID alias case sensitivity and update column mapping in InputsBuilder --- .../core/process/builders/inputs_builder.py | 4 +- backend/core/test/test_inputs_builder.py | 137 ++++++++++++++++++ 2 files changed, 139 insertions(+), 2 deletions(-) create mode 100644 backend/core/test/test_inputs_builder.py diff --git a/backend/core/process/builders/inputs_builder.py b/backend/core/process/builders/inputs_builder.py index dfb14f7..8a1d9a2 100644 --- a/backend/core/process/builders/inputs_builder.py +++ b/backend/core/process/builders/inputs_builder.py @@ -12,7 +12,7 @@ class InputsBuilder: "ra": "RA", "dec": "Dec", "z": "z", - "id": "id", + "id": "ID", "z_flag": "z_flag", "z_err": "z_err", "survey": "survey", @@ -76,7 +76,7 @@ def _build_columns(self, product): return columns def _get_column(self, product, alias): - matches = product.contents.filter(alias=alias) + matches = product.contents.filter(alias__iexact=alias) if matches.count() != 1: LOGGER.warning( diff --git a/backend/core/test/test_inputs_builder.py b/backend/core/test/test_inputs_builder.py new file mode 100644 index 0000000..e166577 --- /dev/null +++ b/backend/core/test/test_inputs_builder.py @@ -0,0 +1,137 @@ +from core.models import Product, ProductContent, ProductType +from core.process.builders.inputs_builder import InputsBuilder +from django.contrib.auth.models import User +from django.test import TestCase + + +class InputsBuilderTestCase(TestCase): + fixtures = [ + "core/fixtures/initial_data.yaml", + ] + + def setUp(self): + self.user = User.objects.create_user( + "john", "john@snow.com", "you_know_nothing" + ) + self.product_type = ProductType.objects.get(name="redshift_catalog") + self.product = Product.objects.create( + product_type=self.product_type, + user=self.user, + display_name="Sample Redshift Catalog", + official_product=False, + path="redshift_catalog/sample", + internal_name="sample_redshift_catalog", + ) + + def test_build_columns_uses_id_alias_from_ucd_mapping(self): + ProductContent.objects.bulk_create([ + ProductContent( + product=self.product, + column_name="object_id", + ucd="meta.id;meta.main", + alias="ID", + order=0, + ), + ProductContent( + product=self.product, + column_name="RAdeg", + ucd="pos.eq.ra;meta.main", + alias="RA", + order=1, + ), + ProductContent( + product=self.product, + column_name="DEdeg", + ucd="pos.eq.dec;meta.main", + alias="Dec", + order=2, + ), + ProductContent( + product=self.product, + column_name="zfinal", + ucd="src.redshift", + alias="z", + order=3, + ), + ]) + + columns = InputsBuilder(process=None)._build_columns(self.product) + + self.assertEqual(columns["id"], "object_id") + self.assertEqual(columns["ra"], "RAdeg") + self.assertEqual(columns["dec"], "DEdeg") + self.assertEqual(columns["z"], "zfinal") + + def test_build_columns_accepts_id_alias_case_variations(self): + ProductContent.objects.bulk_create([ + ProductContent( + product=self.product, + column_name="object_id", + ucd="meta.id;meta.main", + alias="Id", + order=0, + ), + ProductContent( + product=self.product, + column_name="RAdeg", + ucd="pos.eq.ra;meta.main", + alias="RA", + order=1, + ), + ProductContent( + product=self.product, + column_name="DEdeg", + ucd="pos.eq.dec;meta.main", + alias="Dec", + order=2, + ), + ProductContent( + product=self.product, + column_name="zfinal", + ucd="src.redshift", + alias="z", + order=3, + ), + ]) + + columns = InputsBuilder(process=None)._build_columns(self.product) + + self.assertEqual(columns["id"], "object_id") + + def test_build_columns_accepts_required_alias_case_variations(self): + ProductContent.objects.bulk_create([ + ProductContent( + product=self.product, + column_name="object_id", + ucd="meta.id;meta.main", + alias="ID", + order=0, + ), + ProductContent( + product=self.product, + column_name="RAdeg", + ucd="pos.eq.ra;meta.main", + alias="ra", + order=1, + ), + ProductContent( + product=self.product, + column_name="DEdeg", + ucd="pos.eq.dec;meta.main", + alias="DEC", + order=2, + ), + ProductContent( + product=self.product, + column_name="zfinal", + ucd="src.redshift", + alias="Z", + order=3, + ), + ]) + + columns = InputsBuilder(process=None)._build_columns(self.product) + + self.assertEqual(columns["ra"], "RAdeg") + self.assertEqual(columns["dec"], "DEdeg") + self.assertEqual(columns["z"], "zfinal") From 700eeb84b73e9cb4ca6ffdf75a3114313c16aa98 Mon Sep 17 00:00:00 2001 From: Cristiano Singulani Date: Thu, 25 Jun 2026 11:33:39 -0300 Subject: [PATCH 3/4] Add is_directory attribute to main_file in ProductViewSet --- backend/core/views/product.py | 1 + 1 file changed, 1 insertion(+) diff --git a/backend/core/views/product.py b/backend/core/views/product.py index db8bffb..9de5db2 100644 --- a/backend/core/views/product.py +++ b/backend/core/views/product.py @@ -607,6 +607,7 @@ def main_file_info(self, request, **kwargs): main_file["extension"] = product_file.extension main_file["size"] = product_file.size main_file["n_rows"] = product_file.n_rows + main_file["is_directory"] = product_file.is_directory product_contents = self.__get_product_contents(product) if product_contents: From 71ffd54a35fb3b707ef826bce37891d9753a0545 Mon Sep 17 00:00:00 2001 From: Cristiano Singulani Date: Fri, 26 Jun 2026 13:47:50 -0300 Subject: [PATCH 4/4] Add main file archive preparation and download functionality - Implemented `prepare_main_file_archive` method in `ProductDownloadArchiveService` to handle caching and preparing main file downloads. - Created `build_product_main_file_archive` task to manage the asynchronous creation of main file archives. - Updated `ProductViewSet` to include endpoints for preparing and downloading main file archives. - Enhanced tests to cover new main file archive functionality, including caching and directory zipping. - Modified Nginx configuration files to include timeout settings for uwsgi. --- backend/core/services/product_download.py | 197 +++++++++++- backend/core/tasks.py | 60 ++++ backend/core/test/test_product_download.py | 352 +++++++++++++++++++-- backend/core/views/product.py | 295 ++++++++++++----- nginx_development-orch.conf | 2 + nginx_development.conf | 2 + nginx_production.conf | 2 + 7 files changed, 801 insertions(+), 109 deletions(-) diff --git a/backend/core/services/product_download.py b/backend/core/services/product_download.py index a75d532..fb16135 100644 --- a/backend/core/services/product_download.py +++ b/backend/core/services/product_download.py @@ -1,14 +1,22 @@ import hashlib import json +import os import pathlib import secrets +import shutil import time import zipfile from datetime import timedelta from urllib.parse import quote, urlencode import yaml -from core.models import Product, ProductDownloadArchiveStatus, ProductStatus +from core.models import ( + FileRoles, + Product, + ProductDownloadArchiveStatus, + ProductFile, + ProductStatus, +) from django.conf import settings from django.core import signing from django.utils import timezone @@ -60,6 +68,38 @@ def prepare_archive(cls, archive): return archive + @classmethod + def prepare_main_file_archive(cls, archive): + product = archive.product + source_signature = cls.build_main_file_source_signature(product) + download_path = cls.get_or_create_main_file_download(product) + archive_path = cls.get_archive_relative_path(download_path) + now = timezone.now() + + archive.status = ProductDownloadArchiveStatus.READY + archive.archive_path = archive_path + archive.filename = download_path.name + archive.size = download_path.stat().st_size + archive.checksum = cls.build_file_checksum(download_path) + archive.source_signature = source_signature + archive.source_updated_at = now + archive.error_message = None + archive.save( + update_fields=[ + "status", + "archive_path", + "filename", + "size", + "checksum", + "source_signature", + "source_updated_at", + "error_message", + "updated_at", + ] + ) + + return archive + @classmethod def build_zip(cls, internal_name, product_path, metadata, output_dir): product_path = pathlib.Path(product_path) @@ -117,6 +157,7 @@ def build_product_metadata(cls, product): "extension": product_file.extension, "size": product_file.size, "n_rows": product_file.n_rows, + "is_directory": product_file.is_directory, } if product_file.role == 0: @@ -172,6 +213,21 @@ def build_source_signature(cls, product_path): return hashlib.sha256(payload.encode("utf-8")).hexdigest() + @classmethod + def build_main_file_source_signature(cls, product): + main_file = cls.get_product_main_file(product) + payload = json.dumps( + { + "scope": "main_file", + "file": main_file.file.name, + "is_directory": main_file.is_directory, + }, + sort_keys=True, + separators=(",", ":"), + ) + + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + @classmethod def build_file_checksum(cls, file_path): digest = hashlib.sha256() @@ -254,15 +310,30 @@ def cleanup_obsolete_archives(cls): @classmethod def cleanup_product_archives(cls, product): product_path = pathlib.Path(settings.MEDIA_ROOT, product.path) - current_signature = None - keep_ready_id = None + current_signatures = set() + keep_ready_ids = set() if product_path.exists(): current_signature = cls.build_source_signature(product_path) + current_signatures.add(current_signature) ready_archive = cls.find_non_expired_ready_archive( product, current_signature ) - keep_ready_id = ready_archive.pk if ready_archive else None + if ready_archive: + keep_ready_ids.add(ready_archive.pk) + + try: + main_file_signature = cls.build_main_file_source_signature(product) + except ProductFile.DoesNotExist: + main_file_signature = None + + if main_file_signature: + current_signatures.add(main_file_signature) + ready_archive = cls.find_non_expired_ready_archive( + product, main_file_signature + ) + if ready_archive: + keep_ready_ids.add(ready_archive.pk) totals = { "deleted_archives": 0, @@ -270,7 +341,7 @@ def cleanup_product_archives(cls, product): } for archive in product.download_archives.all(): - if cls.should_delete_archive(archive, current_signature, keep_ready_id): + if cls.should_delete_archive(archive, current_signatures, keep_ready_ids): deleted_file = cls.delete_archive_file(archive) archive.delete() totals["deleted_archives"] += 1 @@ -279,14 +350,16 @@ def cleanup_product_archives(cls, product): return totals @classmethod - def should_delete_archive(cls, archive, current_signature, keep_ready_id=None): - if current_signature is None: + def should_delete_archive(cls, archive, current_signatures, keep_ready_ids=None): + keep_ready_ids = keep_ready_ids or set() + + if not current_signatures: return True if archive.status == ProductDownloadArchiveStatus.FAILED: return True - if archive.source_signature != current_signature: + if archive.source_signature not in current_signatures: return True if archive.status == ProductDownloadArchiveStatus.READY: @@ -296,7 +369,7 @@ def should_delete_archive(cls, archive, current_signature, keep_ready_id=None): if cls.is_archive_expired(archive): return True - return keep_ready_id is not None and archive.pk != keep_ready_id + return keep_ready_ids and archive.pk not in keep_ready_ids return False @@ -328,12 +401,103 @@ def get_archive_file_path(cls, archive): @classmethod def get_archive_internal_redirect_path(cls, archive): archive_file_path = cls.get_archive_file_path(archive).resolve() + return cls.get_download_internal_redirect_path(archive_file_path) + + @classmethod + def get_download_internal_redirect_path(cls, file_path): + file_path = pathlib.Path(file_path).resolve() archive_root = cls.get_archive_root().resolve() - relative_path = archive_file_path.relative_to(archive_root).as_posix() + relative_path = file_path.relative_to(archive_root).as_posix() internal_url = cls.get_archive_internal_url().rstrip("/") return f"{internal_url}/{quote(relative_path, safe='/')}" + @classmethod + def get_main_file_download_output_dir(cls, product): + return cls.get_archive_root() / "main-files" / str(product.pk) + + @classmethod + def get_main_file_download_path(cls, product, main_file_path=None): + main_file_path = main_file_path or cls.get_main_file_path(product) + main_file_path = pathlib.Path(main_file_path) + suffix = ".zip" if main_file_path.is_dir() else "" + filename = f"{main_file_path.name}{suffix}" + return cls.get_main_file_download_output_dir(product) / filename + + @classmethod + def get_or_create_main_file_download(cls, product, main_file_path=None): + main_file_path = main_file_path or cls.get_main_file_path(product) + main_file_path = pathlib.Path(main_file_path) + download_path = cls.get_main_file_download_path(product, main_file_path) + + if download_path.is_file(): + return download_path + + download_path.parent.mkdir(parents=True, exist_ok=True) + + if main_file_path.is_dir(): + return cls.build_directory_zip(main_file_path, download_path) + + return cls.cache_main_file(main_file_path, download_path) + + @classmethod + def get_product_main_file(cls, product): + return product.files.get(role=FileRoles.MAIN) + + @classmethod + def get_main_file_path(cls, product): + main_file = cls.get_product_main_file(product) + return pathlib.Path(main_file.file.path) + + @classmethod + def cache_main_file(cls, source_path, download_path): + source_path = pathlib.Path(source_path) + download_path = pathlib.Path(download_path) + + try: + os.link(source_path, download_path) + return download_path + except FileExistsError: + return download_path + except OSError: + pass + + temporary_download_path = download_path.with_name( + f".{download_path.name}.{secrets.token_hex(8)}.tmp" + ) + + try: + shutil.copy2(source_path, temporary_download_path) + temporary_download_path.replace(download_path) + finally: + temporary_download_path.unlink(missing_ok=True) + + return download_path + + @classmethod + def build_directory_zip(cls, directory_path, zip_path): + directory_path = pathlib.Path(directory_path) + zip_path = pathlib.Path(zip_path) + temporary_zip_path = zip_path.with_name( + f".{zip_path.name}.{secrets.token_hex(8)}.tmp" + ) + + try: + with zipfile.ZipFile( + temporary_zip_path, + "w", + compression=zipfile.ZIP_DEFLATED, + compresslevel=cls.get_compression_level(), + ) as zip_handle: + for file_path, arcname in cls.iter_source_files(directory_path): + zip_handle.write(file_path, arcname=arcname) + + temporary_zip_path.replace(zip_path) + finally: + temporary_zip_path.unlink(missing_ok=True) + + return zip_path + @classmethod def build_download_url(cls, archive, user, request=None): token = cls.build_download_token(archive, user) @@ -347,6 +511,19 @@ def build_download_url(cls, archive, user, request=None): return path + @classmethod + def build_main_file_download_url(cls, archive, user, request=None): + token = cls.build_download_token(archive, user) + path = ( + f"/api/products/{archive.product_id}/download/main-file/file/" + f"?{urlencode({'token': token})}" + ) + + if request: + return request.build_absolute_uri(path) + + return path + @classmethod def build_download_token(cls, archive, user): return signing.dumps( diff --git a/backend/core/tasks.py b/backend/core/tasks.py index 04eccc1..d6a0b64 100644 --- a/backend/core/tasks.py +++ b/backend/core/tasks.py @@ -80,6 +80,66 @@ def build_product_download_archive(self, archive_id): } +@shared_task(bind=True) +def build_product_main_file_archive(self, archive_id): + """Builds/caches a product main-file download artifact.""" + started_at = time.monotonic() + archive = ProductDownloadArchive.objects.select_related( + "product", + "product__product_type", + "product__release", + "product__user", + ).get(pk=archive_id) + task_id = self.request.id + + archive.status = ProductDownloadArchiveStatus.RUNNING + archive.task_id = task_id + archive.error_message = None + archive.save( + update_fields=["status", "task_id", "error_message", "updated_at"] + ) + LOGGER.info( + "build_product_main_file_archive started task_id=%s archive_id=%s product_id=%s", + task_id, + archive.pk, + archive.product_id, + ) + + try: + archive = ProductDownloadArchiveService.prepare_main_file_archive(archive) + except Exception as error: + elapsed = time.monotonic() - started_at + LOGGER.exception( + "build_product_main_file_archive failed task_id=%s archive_id=%s product_id=%s elapsed_seconds=%.3f", + task_id, + archive_id, + archive.product_id, + elapsed, + ) + archive.status = ProductDownloadArchiveStatus.FAILED + archive.error_message = str(error) + archive.save(update_fields=["status", "error_message", "updated_at"]) + raise + + elapsed = time.monotonic() - started_at + LOGGER.info( + "build_product_main_file_archive finished task_id=%s archive_id=%s product_id=%s elapsed_seconds=%.3f archive_path=%s size=%s", + task_id, + archive.pk, + archive.product_id, + elapsed, + archive.archive_path, + archive.size, + ) + + return { + "archive_id": archive.pk, + "task_id": task_id, + "status": archive.status, + "archive_path": archive.archive_path, + } + + @shared_task(bind=True) def build_product_table_preview(self, product_id): """Builds product table preview in background and clears processing marker.""" diff --git a/backend/core/test/test_product_download.py b/backend/core/test/test_product_download.py index 72d8c4b..bd18253 100644 --- a/backend/core/test/test_product_download.py +++ b/backend/core/test/test_product_download.py @@ -15,7 +15,11 @@ Release, ) from core.services.product_download import ProductDownloadArchiveService -from core.tasks import build_product_download_archive, cleanup_product_download_archives +from core.tasks import ( + build_product_download_archive, + build_product_main_file_archive, + cleanup_product_download_archives, +) from django.contrib.auth.models import Group, User from django.test import TestCase, override_settings from django.utils import timezone @@ -428,6 +432,111 @@ def test_build_product_download_archive_task_marks_archive_ready(self): self.assertIn("data.csv", zip_file.namelist()) self.assertIn("product_metadata.yaml", zip_file.namelist()) + def test_build_product_main_file_archive_task_caches_regular_file(self): + with tempfile.TemporaryDirectory() as media_root: + download_root = Path(media_root) / "downloads" + product_path = Path(media_root) / self.product.path + product_path.mkdir(parents=True) + data_path = product_path / "data.csv" + data_path.write_text("a,b\n1,2\n", encoding="utf-8") + + ProductFile.objects.create( + product=self.product, + role=FileRoles.MAIN, + name="data.csv", + type="text/csv", + extension=".csv", + size=data_path.stat().st_size, + file=f"{self.product.path}/data.csv", + is_directory=False, + ) + + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=download_root, + ): + archive = ProductDownloadArchive.objects.create( + product=self.product, + created_by=self.user, + filename="data.csv", + source_signature="0" * 64, + ) + + result = build_product_main_file_archive(archive.pk) + + archive.refresh_from_db() + archive_path = Path(media_root) / archive.archive_path + cached_content = archive_path.read_text(encoding="utf-8") + + self.assertEqual(result["archive_id"], archive.pk) + self.assertEqual(archive.status, ProductDownloadArchiveStatus.READY) + self.assertEqual(archive.filename, "data.csv") + self.assertEqual(cached_content, "a,b\n1,2\n") + self.assertEqual( + archive.archive_path, + f"downloads/main-files/{self.product.pk}/data.csv", + ) + self.assertEqual(len(archive.checksum), 64) + self.assertNotEqual(archive.source_signature, "0" * 64) + + def test_build_product_main_file_archive_task_zips_directory(self): + with tempfile.TemporaryDirectory() as media_root: + download_root = Path(media_root) / "downloads" + product_path = Path(media_root) / self.product.path + hats_path = product_path / "hats_catalog" + data_path = hats_path / "dataset" / "Norder=0" / "Dir=0" + data_path.mkdir(parents=True) + (hats_path / "collection.properties").write_text( + "catalog_name=hats_catalog\n", + encoding="utf-8", + ) + (data_path / "Npix=0.parquet").write_bytes(b"parquet") + + ProductFile.objects.create( + product=self.product, + role=FileRoles.MAIN, + name="hats_catalog", + type="application/x-hats", + extension="", + size=0, + file=f"{self.product.path}/hats_catalog", + is_directory=True, + ) + + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=download_root, + ): + archive = ProductDownloadArchive.objects.create( + product=self.product, + created_by=self.user, + filename="hats_catalog.zip", + source_signature="0" * 64, + ) + + result = build_product_main_file_archive(archive.pk) + + archive.refresh_from_db() + archive_path = Path(media_root) / archive.archive_path + + with zipfile.ZipFile(archive_path) as zip_file: + names = sorted(zip_file.namelist()) + + self.assertEqual(result["archive_id"], archive.pk) + self.assertEqual(archive.status, ProductDownloadArchiveStatus.READY) + self.assertEqual(archive.filename, "hats_catalog.zip") + self.assertEqual( + archive.archive_path, + f"downloads/main-files/{self.product.pk}/hats_catalog.zip", + ) + self.assertEqual( + names, + [ + "collection.properties", + "dataset/Norder=0/Dir=0/Npix=0.parquet", + ], + ) + def test_build_product_download_archive_task_marks_archive_failed(self): archive = ProductDownloadArchive.objects.create( product=self.product, @@ -488,6 +597,23 @@ def create_product_source_file(self, media_root): (product_path / "data.csv").write_text("a,b\n1,2\n", encoding="utf-8") return product_path + def create_product_main_file(self, media_root): + product_path = self.create_product_source_file(media_root) + data_path = product_path / "data.csv" + + ProductFile.objects.create( + product=self.product, + role=FileRoles.MAIN, + name="data.csv", + type="text/csv", + extension=".csv", + size=data_path.stat().st_size, + file=f"{self.product.path}/data.csv", + is_directory=False, + ) + + return data_path + def create_hats_main_file(self, media_root): product_path = Path(media_root) / self.product.path hats_path = product_path / "hats_catalog" @@ -882,37 +1008,221 @@ def test_legacy_download_endpoint_includes_deprecation_headers(self): response["Link"], ) - def test_download_main_file_returns_zip_for_hats_directory(self): + def test_download_main_file_prepare_caches_regular_file_synchronously(self): + with tempfile.TemporaryDirectory() as media_root: + self.create_product_main_file(media_root) + + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=Path(media_root) / "downloads", + PRODUCT_DOWNLOAD_INTERNAL_URL="/internal-downloads/", + PRODUCT_DOWNLOAD_PREPARE_WAIT_SECONDS=0, + ): + source_signature = ( + ProductDownloadArchiveService.build_main_file_source_signature( + self.product + ) + ) + with mock.patch( + "core.views.product.build_product_main_file_archive.delay" + ) as delay_mock: + response = self.client.post( + f"/api/products/{self.product.pk}/download/main-file/prepare/" + ) + cached_path = ( + Path(media_root) + / "downloads" + / "main-files" + / str(self.product.pk) + / "data.csv" + ) + cached_content = cached_path.read_text(encoding="utf-8") + + archive = ProductDownloadArchive.objects.get(product=self.product) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data["id"], archive.pk) + self.assertEqual(response.data["status"], ProductDownloadArchiveStatus.READY) + self.assertIsNone(archive.task_id) + self.assertEqual(archive.filename, "data.csv") + self.assertEqual(archive.source_signature, source_signature) + self.assertEqual(cached_content, "a,b\n1,2\n") + self.assertIn("/download/main-file/file/", response.data["download_url"]) + delay_mock.assert_not_called() + + def test_download_main_file_prepare_queues_task_for_hats_directory(self): with tempfile.TemporaryDirectory() as media_root: self.create_hats_main_file(media_root) + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=Path(media_root) / "downloads", + PRODUCT_DOWNLOAD_INTERNAL_URL="/internal-downloads/", + PRODUCT_DOWNLOAD_PREPARE_WAIT_SECONDS=0, + ): + source_signature = ( + ProductDownloadArchiveService.build_main_file_source_signature( + self.product + ) + ) + with mock.patch( + "core.views.product.build_product_main_file_archive.delay" + ) as delay_mock: + delay_mock.return_value.id = "main-task-1" + + response = self.client.post( + f"/api/products/{self.product.pk}/download/main-file/prepare/" + ) + + archive = ProductDownloadArchive.objects.get(product=self.product) + + self.assertEqual(response.status_code, 202) + self.assertEqual(response.data["id"], archive.pk) + self.assertEqual(response.data["status"], ProductDownloadArchiveStatus.PENDING) + self.assertEqual(archive.task_id, "main-task-1") + self.assertEqual(archive.filename, "hats_catalog.zip") + self.assertEqual(archive.source_signature, source_signature) + delay_mock.assert_called_once_with(archive.pk) + + def test_download_main_file_prepare_reuses_ready_archive(self): + with tempfile.TemporaryDirectory() as media_root: + self.create_product_main_file(media_root) + archive_relative_path = Path( + f"downloads/main-files/{self.product.pk}/data.csv" + ) + archive_path = Path(media_root) / archive_relative_path + archive_path.parent.mkdir(parents=True) + archive_path.write_text("a,b\n1,2\n", encoding="utf-8") + + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=Path(media_root) / "downloads", + PRODUCT_DOWNLOAD_INTERNAL_URL="/internal-downloads/", + ): + source_signature = ( + ProductDownloadArchiveService.build_main_file_source_signature( + self.product + ) + ) + archive = ProductDownloadArchive.objects.create( + product=self.product, + created_by=self.user, + status=ProductDownloadArchiveStatus.READY, + archive_path=archive_relative_path.as_posix(), + filename="data.csv", + size=archive_path.stat().st_size, + checksum="a" * 64, + source_signature=source_signature, + ) + + with mock.patch( + "core.views.product.build_product_main_file_archive.delay" + ) as delay_mock: + response = self.client.post( + f"/api/products/{self.product.pk}/download/main-file/prepare/" + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data["id"], archive.pk) + self.assertEqual(response.data["status"], ProductDownloadArchiveStatus.READY) + self.assertIn("/download/main-file/file/", response.data["download_url"]) + delay_mock.assert_not_called() + + def test_download_main_file_status_returns_not_found_when_no_archive_matches(self): + with tempfile.TemporaryDirectory() as media_root: + self.create_product_main_file(media_root) + with override_settings(MEDIA_ROOT=media_root): response = self.client.get( - f"/api/products/{self.product.pk}/download_main_file/" + f"/api/products/{self.product.pk}/download/main-file/status/" ) - content = b"".join(response.streaming_content) + self.assertEqual(response.status_code, 404) + self.assertEqual(response.data["status"], "not_found") + + def test_download_main_file_file_uses_internal_redirect_for_regular_file(self): + with tempfile.TemporaryDirectory() as media_root: + data_path = self.create_product_main_file(media_root) + data_size = data_path.stat().st_size + archive_relative_path = Path( + f"downloads/main-files/{self.product.pk}/data.csv" + ) + archive_path = Path(media_root) / archive_relative_path + archive_path.parent.mkdir(parents=True) + archive_path.write_text( + data_path.read_text(encoding="utf-8"), + encoding="utf-8", + ) + + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=Path(media_root) / "downloads", + PRODUCT_DOWNLOAD_INTERNAL_URL="/internal-downloads/", + ): + source_signature = ( + ProductDownloadArchiveService.build_main_file_source_signature( + self.product + ) + ) + archive = ProductDownloadArchive.objects.create( + product=self.product, + created_by=self.user, + status=ProductDownloadArchiveStatus.READY, + archive_path=archive_relative_path.as_posix(), + filename="data.csv", + size=archive_path.stat().st_size, + checksum="a" * 64, + source_signature=source_signature, + ) + token = ProductDownloadArchiveService.build_download_token( + archive, self.user + ) + response = self.client.get( + f"/api/products/{self.product.pk}/download/main-file/file/", + {"token": token}, + ) self.assertEqual(response.status_code, 200) - self.assertEqual(response["Content-Type"], "application/zip") - self.assertIn( - 'filename="hats_catalog.zip"', + self.assertEqual(response["Content-Length"], str(data_size)) + self.assertEqual( response["Content-Disposition"], + "attachment; filename=data.csv", + ) + self.assertEqual( + response["X-Accel-Redirect"], + f"/internal-downloads/main-files/{self.product.pk}/data.csv", ) - with tempfile.TemporaryDirectory() as tmpdirname: - zip_path = Path(tmpdirname) / "hats_catalog.zip" - zip_path.write_bytes(content) + def test_download_main_file_legacy_endpoint_returns_accepted_while_task_runs(self): + with tempfile.TemporaryDirectory() as media_root: + self.create_hats_main_file(media_root) - with zipfile.ZipFile(zip_path) as archive: - self.assertEqual( - sorted(archive.namelist()), - [ - "collection.properties", - "dataset/Norder=0/Dir=0/Npix=0.parquet", - ], - ) - self.assertEqual( - archive.read("collection.properties").decode("utf-8"), - "catalog_name=hats_catalog\n", + with override_settings( + MEDIA_ROOT=media_root, + PRODUCT_DOWNLOAD_ROOT=Path(media_root) / "downloads", + PRODUCT_DOWNLOAD_INTERNAL_URL="/internal-downloads/", + PRODUCT_DOWNLOAD_PREPARE_WAIT_SECONDS=0, + ): + zip_path = ( + Path(media_root) + / "downloads" + / "main-files" + / str(self.product.pk) + / "hats_catalog.zip" ) + with mock.patch( + "core.views.product.build_product_main_file_archive.delay" + ) as delay_mock: + delay_mock.return_value.id = "main-task-1" + + response = self.client.get( + f"/api/products/{self.product.pk}/download_main_file/" + ) + zip_exists = zip_path.exists() + + archive = ProductDownloadArchive.objects.get(product=self.product) + self.assertEqual(response.status_code, 202) + self.assertEqual(response.data["id"], archive.pk) + self.assertEqual(response.data["status"], ProductDownloadArchiveStatus.PENDING) + self.assertFalse(zip_exists) + delay_mock.assert_called_once_with(archive.pk) diff --git a/backend/core/views/product.py b/backend/core/views/product.py index 9de5db2..81a281b 100644 --- a/backend/core/views/product.py +++ b/backend/core/views/product.py @@ -2,7 +2,6 @@ import mimetypes import pathlib import tempfile -import zipfile from json import dumps, loads from pathlib import Path @@ -19,7 +18,11 @@ from core.product_steps import CreateProduct, NonAdminError, RegistryProduct from core.serializers import ProductSerializer from core.services import AccessControlService, ProductDownloadArchiveService -from core.tasks import build_product_download_archive, build_product_table_preview +from core.tasks import ( + build_product_download_archive, + build_product_main_file_archive, + build_product_table_preview, +) from core.utils import format_query_to_char from django.conf import settings from django.core import signing @@ -211,6 +214,7 @@ def __get_full_product(self, product): "extension": file.extension, "size": file.size, "n_rows": file.n_rows, + "is_directory": file.is_directory, } product_contents = self.__get_product_contents(product) if product_contents: @@ -224,32 +228,6 @@ def __get_full_product(self, product): return product_full - def __build_directory_zip_response(self, directory_path): - zip_handle = tempfile.TemporaryFile() - - with zipfile.ZipFile( - zip_handle, - "w", - compression=zipfile.ZIP_DEFLATED, - compresslevel=ProductDownloadArchiveService.get_compression_level(), - ) as zip_file: - for file_path in sorted(directory_path.rglob("*")): - if not file_path.is_file(): - continue - - arcname = file_path.relative_to(directory_path).as_posix() - zip_file.write(file_path, arcname=arcname) - - size = zip_handle.tell() - zip_handle.seek(0) - - response = FileResponse(zip_handle, content_type="application/zip") - response["Content-Length"] = size - response["Content-Disposition"] = ( - f'attachment; filename="{directory_path.name}.zip"' - ) - return response - def __get_flag_translation_name(self, config): """Extract the flags translation filename from a nested config.""" if isinstance(config, dict): @@ -297,7 +275,7 @@ def process_config(self, request, **kwargs): return Response(response) - def __serialize_download_archive(self, archive, request): + def __serialize_download_archive(self, archive, request, main_file=False): data = { "id": archive.pk, "task_id": archive.task_id, @@ -311,11 +289,20 @@ def __serialize_download_archive(self, archive, request): } if archive.status == ProductDownloadArchiveStatus.READY: - data["download_url"] = ProductDownloadArchiveService.build_download_url( - archive, - request.user, - request, - ) + if main_file: + data["download_url"] = ( + ProductDownloadArchiveService.build_main_file_download_url( + archive, + request.user, + request, + ) + ) + else: + data["download_url"] = ProductDownloadArchiveService.build_download_url( + archive, + request.user, + request, + ) data["expired_time"] = ( ProductDownloadArchiveService.get_archive_expired_time(archive) ) @@ -329,6 +316,73 @@ def __get_current_download_signature(self, product): product_path = pathlib.Path(settings.MEDIA_ROOT, product.path) return ProductDownloadArchiveService.build_source_signature(product_path) + def __get_current_main_file_signature(self, product): + return ProductDownloadArchiveService.build_main_file_source_signature(product) + + def __build_archive_file_response(self, archive): + archive_path = ProductDownloadArchiveService.get_archive_file_path(archive) + if not archive_path.exists(): + return Response( + {"error": "Download archive file not found."}, + status=status.HTTP_404_NOT_FOUND, + ) + + mimetype, _ = mimetypes.guess_type(archive_path) + response = HttpResponse(content_type=mimetype) + response["X-Accel-Redirect"] = ( + ProductDownloadArchiveService.get_archive_internal_redirect_path(archive) + ) + response["Content-Length"] = archive_path.stat().st_size + response["Content-Disposition"] = "attachment; filename={}".format( + archive.filename + ) + return response + + def __load_ready_archive_from_token(self, product, token): + if not token: + raise exceptions.PermissionDenied("Missing download token.") + + try: + payload = ProductDownloadArchiveService.load_download_token(token) + except signing.SignatureExpired: + raise exceptions.PermissionDenied("Download token has expired.") + except signing.BadSignature: + raise exceptions.PermissionDenied("Invalid download token.") + + if str(payload.get("product_id")) != str(product.pk): + raise exceptions.PermissionDenied("Invalid download token.") + + return ProductDownloadArchive.objects.select_related("product").get( + pk=payload.get("archive_id"), + product_id=product.pk, + status=ProductDownloadArchiveStatus.READY, + ) + + def __prepare_or_queue_main_file_archive(self, archive, product, log_context): + if ProductDownloadArchiveService.get_main_file_path(product).is_dir(): + task = build_product_main_file_archive.delay(archive.pk) + archive.task_id = task.id + archive.save(update_fields=["task_id", "updated_at"]) + logger.info( + "%s queued archive_id=%s task_id=%s product_id=%s source_signature=%s", + log_context, + archive.pk, + archive.task_id, + product.pk, + archive.source_signature, + ) + return ProductDownloadArchiveService.wait_for_archive_preparation(archive) + + archive = ProductDownloadArchiveService.prepare_main_file_archive(archive) + logger.info( + "%s prepared synchronously archive_id=%s product_id=%s source_signature=%s", + log_context, + archive.pk, + product.pk, + archive.source_signature, + ) + return archive + @action(methods=["POST"], detail=True, url_path="download/prepare") def download_prepare(self, request, **kwargs): product = self.get_object() @@ -412,25 +466,100 @@ def download_status(self, request, **kwargs): ) def download_file(self, request, **kwargs): product = self.get_object() - token = request.GET.get("token") - if not token: - raise exceptions.PermissionDenied("Missing download token.") - try: - payload = ProductDownloadArchiveService.load_download_token(token) - except signing.SignatureExpired: - raise exceptions.PermissionDenied("Download token has expired.") - except signing.BadSignature: - raise exceptions.PermissionDenied("Invalid download token.") + archive = self.__load_ready_archive_from_token( + product, + request.GET.get("token"), + ) + except ProductDownloadArchive.DoesNotExist: + return Response( + {"error": "Download archive not found."}, + status=status.HTTP_404_NOT_FOUND, + ) - if str(payload.get("product_id")) != str(product.pk): - raise exceptions.PermissionDenied("Invalid download token.") + return self.__build_archive_file_response(archive) + + @action(methods=["POST"], detail=True, url_path="download/main-file/prepare") + def download_main_file_prepare(self, request, **kwargs): + product = self.get_object() + source_signature = self.__get_current_main_file_signature(product) + archive = ProductDownloadArchiveService.find_current_archive( + product, + source_signature, + ) + + if archive and archive.status in ( + ProductDownloadArchiveStatus.PENDING, + ProductDownloadArchiveStatus.RUNNING, + ProductDownloadArchiveStatus.READY, + ): + logger.info( + "download_main_file_prepare reusing archive_id=%s task_id=%s product_id=%s status=%s", + archive.pk, + archive.task_id, + product.pk, + archive.status, + ) + response_status = ( + status.HTTP_200_OK + if archive.status == ProductDownloadArchiveStatus.READY + else status.HTTP_202_ACCEPTED + ) + return Response( + self.__serialize_download_archive(archive, request, main_file=True), + status=response_status, + ) + + archive = ProductDownloadArchive.objects.create( + product=product, + created_by=request.user, + filename=ProductDownloadArchiveService.get_main_file_download_path( + product + ).name, + source_signature=source_signature, + ) + archive = self.__prepare_or_queue_main_file_archive( + archive, + product, + "download_main_file_prepare", + ) + response_status = ( + status.HTTP_200_OK + if archive.status == ProductDownloadArchiveStatus.READY + else status.HTTP_202_ACCEPTED + ) + + return Response( + self.__serialize_download_archive(archive, request, main_file=True), + status=response_status, + ) + + @action(methods=["GET"], detail=True, url_path="download/main-file/status") + def download_main_file_status(self, request, **kwargs): + product = self.get_object() + source_signature = self.__get_current_main_file_signature(product) + archive = ProductDownloadArchiveService.find_current_archive( + product, + source_signature, + ) + + if not archive: + return Response( + {"status": "not_found"}, + status=status.HTTP_404_NOT_FOUND, + ) + + return Response( + self.__serialize_download_archive(archive, request, main_file=True) + ) + @action(methods=["GET"], detail=True, url_path="download/main-file/file") + def download_main_file_file(self, request, **kwargs): + product = self.get_object() try: - archive = ProductDownloadArchive.objects.select_related("product").get( - pk=payload.get("archive_id"), - product_id=product.pk, - status=ProductDownloadArchiveStatus.READY, + archive = self.__load_ready_archive_from_token( + product, + request.GET.get("token"), ) except ProductDownloadArchive.DoesNotExist: return Response( @@ -438,23 +567,13 @@ def download_file(self, request, **kwargs): status=status.HTTP_404_NOT_FOUND, ) - archive_path = ProductDownloadArchiveService.get_archive_file_path(archive) - if not archive_path.exists(): + if archive.source_signature != self.__get_current_main_file_signature(product): return Response( - {"error": "Download archive file not found."}, + {"error": "Download archive not found."}, status=status.HTTP_404_NOT_FOUND, ) - mimetype, _ = mimetypes.guess_type(archive_path) - response = HttpResponse(content_type=mimetype) - response["X-Accel-Redirect"] = ( - ProductDownloadArchiveService.get_archive_internal_redirect_path(archive) - ) - response["Content-Length"] = archive_path.stat().st_size - response["Content-Disposition"] = "attachment; filename={}".format( - archive.filename - ) - return response + return self.__build_archive_file_response(archive) @action(methods=["GET"], detail=True) def download(self, request, **kwargs): @@ -494,26 +613,46 @@ def download_main_file(self, request, **kwargs): """Download product main file""" try: product = self.get_object() - main_file = product.files.get(role=0) - main_file_path = Path(main_file.file.path) - product_path = pathlib.Path( - settings.MEDIA_ROOT, product.path, main_file_path + source_signature = self.__get_current_main_file_signature(product) + archive = ProductDownloadArchiveService.find_current_archive( + product, + source_signature, ) - if product_path.is_dir(): - return self.__build_directory_zip_response(product_path) - - # Abre o arquivo e envia em bites para o navegador - mimetype, _ = mimetypes.guess_type(product_path) - size = product_path.stat().st_size - name = product_path.name + if archive and archive.status == ProductDownloadArchiveStatus.READY: + return self.__build_archive_file_response(archive) + + if not archive or archive.status == ProductDownloadArchiveStatus.FAILED: + archive = ProductDownloadArchive.objects.create( + product=product, + created_by=request.user, + filename=ProductDownloadArchiveService.get_main_file_download_path( + product + ).name, + source_signature=source_signature, + ) + archive = self.__prepare_or_queue_main_file_archive( + archive, + product, + "download_main_file", + ) + else: + archive = ProductDownloadArchiveService.wait_for_archive_preparation( + archive + ) - file_handle = open(product_path, "rb") - response = FileResponse(file_handle, content_type=mimetype) + if archive.status == ProductDownloadArchiveStatus.READY: + return self.__build_archive_file_response(archive) - response["Content-Length"] = size - response["Content-Disposition"] = "attachment; filename={}".format(name) - return response + response_status = ( + status.HTTP_500_INTERNAL_SERVER_ERROR + if archive.status == ProductDownloadArchiveStatus.FAILED + else status.HTTP_202_ACCEPTED + ) + return Response( + self.__serialize_download_archive(archive, request, main_file=True), + status=response_status, + ) except Exception as e: content = {"error": str(e)} return Response(content, status=status.HTTP_500_INTERNAL_SERVER_ERROR) diff --git a/nginx_development-orch.conf b/nginx_development-orch.conf index 75e3a18..6a540bc 100644 --- a/nginx_development-orch.conf +++ b/nginx_development-orch.conf @@ -23,6 +23,8 @@ server { proxy_read_timeout 120s; fastcgi_send_timeout 120s; fastcgi_read_timeout 120s; + uwsgi_send_timeout 120s; + uwsgi_read_timeout 120s; # access_log /var/log/nginx/host.access.log main; diff --git a/nginx_development.conf b/nginx_development.conf index 4eb4146..da02eb6 100644 --- a/nginx_development.conf +++ b/nginx_development.conf @@ -23,6 +23,8 @@ server { proxy_read_timeout 120s; fastcgi_send_timeout 120s; fastcgi_read_timeout 120s; + uwsgi_send_timeout 120s; + uwsgi_read_timeout 120s; # access_log /var/log/nginx/host.access.log main; diff --git a/nginx_production.conf b/nginx_production.conf index 91a154d..dabb900 100644 --- a/nginx_production.conf +++ b/nginx_production.conf @@ -25,6 +25,8 @@ server { proxy_read_timeout 1200; fastcgi_send_timeout 1200; fastcgi_read_timeout 1200; + uwsgi_send_timeout 1200; + uwsgi_read_timeout 1200; # access_log /var/log/nginx/host.access.log main;