Improve user feedback on upload error

dashboard_viewer/dashboard_viewer/settings.py

@@ -157,7 +157,15 @@
     "postgresql"
     f"://{DATABASES['achilles']['USER']}:{DATABASES['achilles']['PASSWORD']}"
     f"@{DATABASES['achilles']['HOST']}:{DATABASES['achilles']['PORT']}"
-    f"/{DATABASES['achilles']['NAME']}"
+    f"/{DATABASES['achilles']['NAME']}",
+    pool_pre_ping=True,      # cheap SELECT 1 before handing out a connection
+    pool_recycle=1800,       # don't reuse anything older than 30 min
+    connect_args={
+        "keepalives": 1,
+        "keepalives_idle": 30,
+        "keepalives_interval": 10,
+        "keepalives_count": 5,
+    },
 )
 
 DATABASE_ROUTERS = ["dashboard_viewer.routers.AchillesRouter"]

dashboard_viewer/uploader/file_handler/checks.py

@@ -1,49 +1,31 @@
 import csv
 import hashlib
 import io
+import logging
 import os
 
 import numpy
 import pandas
 from django.conf import settings
 from django.core.cache import caches
-from django.db import connections, transaction, utils
-from redis_rw_lock import RWLock
-
+from django.db import connections
 from materialized_queries_manager.models import MaterializedQuery
+from redis_rw_lock import RWLock
 from uploader.models import AchillesResults, DataSource, UploadHistory
 
+from .errors import (
+    DuplicatedMetadataRow,
+    EqualFileAlreadyUploaded,
+    FileDataCorrupted,
+    InvalidCSVFile,
+    InvalidFieldValue,
+    InvalidFileFormat,
+    MissingFieldValue,
+    TemporaryFailure,
+    translate,
+)
 
-class FileChecksException(Exception):
-    pass
-
-
-class InvalidCSVFile(FileChecksException):
-    pass
-
-
-class InvalidFileFormat(FileChecksException):
-    pass
-
-
-class InvalidFieldValue(FileChecksException):
-    pass
-
-
-class DuplicatedMetadataRow(FileChecksException):
-    pass
-
-
-class MissingFieldValue(FileChecksException):
-    pass
-
-
-class EqualFileAlreadyUploaded(FileChecksException):
-    pass
-
-class FileDataCorrupted(FileChecksException):
-    pass
-
+logger = logging.getLogger(__name__)
 
 def _generate_file_reader(uploaded_file):
     """
@@ -268,63 +250,82 @@ def _get_upload_attr(analysis, stratum):
         return None
     return value
 
+def _calculate_sha256(file_path):
+    """
+    Calculates the SHA-256 checksum of a file in streaming 64 KB chunks.
+    Reads the target file in fixed-size blocks to safely compute hashes
+    for large files without causing Out-Of-Memory (OOM) errors.
+    Args:
+        file_path (str | PathLike): The absolute or relative path to the file on disk.
+    Returns:
+        str | None: The hexadecimal SHA-256 digest string if successful,
+                    or None if an IOError occurs while reading the file.
+    """
+    hasher = hashlib.sha256()
+    try:
+        with open(file_path, "rb") as f:
+            # Iteratively read 64 KB chunks until EOF (empty byte string)
+            # as we might get large files, so it should prevent OOM
+            for chunk in iter(lambda: f.read(65536), b""):
+                hasher.update(chunk)
+        return hasher.hexdigest()
+    except IOError:
+        return None
 
 def check_for_duplicated_files(uploaded_file, data_source_id):
+    """
+    Verifies that an incoming upload is not an exact duplicate of the latest upload.
+    Fetches the most recent successful upload record from `UploadHistory` for the
+    specified data source and compares its SHA-256 checksum against the new file.
+    Args:
+        uploaded_file (FieldFile | File): Django File object for the incoming upload.
+        data_source_id (int): Primary key ID of the target DataSource.
+    Raises:
+        EqualFileAlreadyUploaded: If both files exist and their SHA-256 hashes match.
+    """
     try:
-        # Upload History only stores the succeded files
-
+        # Retrieve the latest successful upload record for this datasource
         pd = UploadHistory.objects.filter(data_source_id=data_source_id).latest()
 
-        data_source_hash = (
-            DataSource.objects.filter(id=data_source_id)
-            .values_list("hash", flat=True)
-            .first()
-        )
-
-        if data_source_hash is not None and bool(pd.uploaded_file):
-            # Go to the path where success files are stored
-
-            latest_file_path = os.path.join(settings.MEDIA_ROOT, pd.uploaded_file.path)
-
-            if os.path.exists(latest_file_path) and os.path.exists(uploaded_file.path):
-                with open(latest_file_path, "rb") as previous_upload_file:
-                    try:
-                        checksum_previous = hashlib.sha256(
-                            previous_upload_file.read()
-                        ).hexdigest()
-
-                    except IOError:
-                        checksum_previous = None
-
-                with open(uploaded_file.path, "rb") as new_upload_file:
-                    try:
-                        checksum_new = hashlib.sha256(
-                            new_upload_file.read()
-                        ).hexdigest()
-
-                    except IOError:
-                        checksum_new = None
-
-                if (
-                    checksum_previous is not None
-                    and checksum_new is not None
-                    and checksum_previous == checksum_new
-                ):
-                    raise EqualFileAlreadyUploaded("File is already in the database")
-
+        # Validate physical existence of both files before hashing
+        if pd.uploaded_file and os.path.exists(pd.uploaded_file.path) and os.path.exists(uploaded_file.path):
+            if _calculate_sha256(pd.uploaded_file.path) == _calculate_sha256(uploaded_file.path):
+                raise EqualFileAlreadyUploaded("This exact file has already been uploaded for this datasource.")
     except UploadHistory.DoesNotExist:
+        # First upload for this datasource; no prior history to compare against
         pass
 
+def _get_mat_view_queries():
+    """Extracts view definitions, redirects to staging table"""
+    all_mat_views = MaterializedQuery.objects.exclude(matviewname__contains="tmp")
+    mat_views = {}
+
+    for mat_view in all_mat_views:
+        tmp_mat_view_name = mat_view.to_dict()["matviewname"] + "_tmp"
+        # To run the mat views (with data) against the "temporary table"
+        # To run for all mat views, as the data source can become with draft equal to true
+        tmp_definition = mat_view.to_dict()["definition"].replace(
+            "achilles_results", "achilles_results_tmp"
+        )
+        mat_views[tmp_mat_view_name] = [
+            tmp_definition,
+        ]
+        # since draft can change with time, we must run the queries for all types of draft, namely with draft = true and draft = false
+        # We normally don't use draft set to true in the queries, otherwise we would need to test it here also
+        if "draft = false" in tmp_definition:
+            mat_views[tmp_mat_view_name].append(
+                tmp_definition.replace("draft = false", "draft = true")
+            )
+    return mat_views
 
-def upload_data_to_tmp_table(data_source_id, file_metadata, pending_upload):
+def validate_data_in_existing_mat_views(data_source_id, file_metadata, pending_upload):
 
     cache = caches["workers_locks"]
+    ctx = {"ds": data_source_id, "upload": pending_upload.id}
 
     with RWLock(
         cache.client.get_client(), "celery_worker_updating", RWLock.WRITE, expire=None
         ):
-
-        # Upload New Data to a "temporary" table
         pending_upload.uploaded_file.seek(0)
 
         reader = pandas.read_csv(
@@ -337,80 +338,80 @@ def upload_data_to_tmp_table(data_source_id, file_metadata, pending_upload):
             chunksize=500,
         )
 
-        all_mat_views = MaterializedQuery.objects.exclude(matviewname__contains="tmp")
-
-        mat_views = {}
-
-        for mat_view in all_mat_views:
-            tmp_mat_view_name = mat_view.to_dict()["matviewname"] + "_tmp"
-
-            # To run the mat views (with data) against the "temporary table"
-            # To run for all mat views, as the data source can become with draft equal to true
-
-            tmp_definition = mat_view.to_dict()["definition"].replace(
-                "achilles_results", "achilles_results_tmp"
-            )
-
-            mat_views[tmp_mat_view_name] = [
-                tmp_definition,
-            ]
-
-            # since draft can change with time, we must run the queries for all types of draft, namely with draft = true and draft = false
-            if "draft = false" in tmp_definition:
-                mat_views[tmp_mat_view_name].append(
-                    tmp_definition.replace("draft = false", "draft = true")
-                )
-
-        # Create "Temporary Upload" table, to store the data being uploaded
-        # Refresh of Materialized views does not allow the refresh in Temporary Tables
-
-        with transaction.atomic(), connections[
-            "achilles"
-        ].cursor() as cursor, settings.ACHILLES_DB_SQLALCHEMY_ENGINE.connect() as pandas_connection, pandas_connection.begin():
-            try:
-                cursor.execute("DROP TABLE IF EXISTS achilles_results_tmp CASCADE")
-                cursor.execute(
-                    "CREATE TABLE IF NOT EXISTS achilles_results_tmp AS SELECT * FROM "
-                    + AchillesResults._meta.db_table
-                    + " WHERE FALSE"
-                )
-                cursor.execute("CREATE SEQUENCE IF NOT EXISTS achilles_results_tmp_seq_id")
-                cursor.execute(
-                    "ALTER TABLE achilles_results_tmp ALTER COLUMN id SET DEFAULT nextval('achilles_results_tmp_seq_id')"
-                )
-                cursor.execute(
-                    "ALTER TABLE achilles_results_tmp ALTER COLUMN id SET NOT NULL"
+        try:
+            mat_views = _get_mat_view_queries()
+            _load_staging_table(reader, data_source_id, logger, ctx)
+            failed = _probe_charts(mat_views, logger, ctx)
+
+            if failed:
+                raise FileDataCorrupted(
+                    "Some charts couldn't be built "
+                    "from this file. The rows loaded correctly but produced values "
+                    "the database can't store or calculate."
                 )
+        finally:
+            _drop_staging_table(logger)
 
-                # Upload data into "Temporary Table", similar structure to the actual upload process
-                for chunk in reader:
-                    chunk = chunk[chunk["stratum_1"].isin(["0"]) == False]
-                    chunk = chunk.assign(data_source_id=data_source_id)
-                    chunk.to_sql(
-                        "achilles_results_tmp",
-                        pandas_connection,
-                        if_exists="append",
-                        index=False,
-                    )
-            except Exception:
-                raise InvalidCSVFile("Error processing the file")
-
-        # Fetch Materialized Views
-        # The option here is to change the definitions of the original mat views and replace
-        # the achilles_results references to achilles_results_tmp
-
-        with transaction.atomic(), connections["achilles"].cursor() as cursor:
-            try:
-                for tmp_mat_view_name in mat_views:  # noqa
-                    for tmp_definition in mat_views[tmp_mat_view_name]:
-                        cursor.execute(
-                            f"CREATE MATERIALIZED VIEW {tmp_mat_view_name} AS {tmp_definition}"
-                        )
-                        cursor.execute(f"DROP MATERIALIZED VIEW {tmp_mat_view_name}")
-            except utils.DataError:
-                cursor.execute("DROP TABLE IF EXISTS achilles_results_tmp CASCADE")
-                raise FileDataCorrupted("Uploaded file is not valid")
-
-        # Delete Temprary Upload data and its dependent (Materialzied Views)
-        with transaction.atomic(), connections["achilles"].cursor() as cursor:
-            cursor.execute("DROP TABLE IF EXISTS achilles_results_tmp CASCADE")
+def _load_staging_table(reader, data_source_id, log, ctx):
+    with connections["achilles"].cursor() as cursor, \
+            settings.ACHILLES_DB_SQLALCHEMY_ENGINE.connect() as pandas_connection, \
+            pandas_connection.begin():
+        try:
+            cursor.execute("DROP TABLE IF EXISTS achilles_results_tmp CASCADE")
+            cursor.execute(
+                "CREATE TABLE IF NOT EXISTS achilles_results_tmp AS SELECT * FROM "
+                + AchillesResults._meta.db_table + " WHERE FALSE"
+            )
+            cursor.execute("CREATE SEQUENCE IF NOT EXISTS achilles_results_tmp_seq_id")
+            cursor.execute(
+                "ALTER TABLE achilles_results_tmp ALTER COLUMN id "
+                "SET DEFAULT nextval('achilles_results_tmp_seq_id')"
+            )
+            cursor.execute("ALTER TABLE achilles_results_tmp ALTER COLUMN id SET NOT NULL")
+
+            rows = 0
+            for chunk in reader:
+                chunk = (chunk[~chunk["stratum_1"].eq("0")]
+                         .assign(data_source_id=data_source_id))
+                chunk.to_sql("achilles_results_tmp", pandas_connection,
+                             if_exists="append", index=False)
+                rows += len(chunk)
+            log.info("Staged %d rows %s", rows, ctx)
+        except Exception as exc:
+            raise translate(exc, "loading your data", log, **ctx) from exc
+
+def _probe_charts(mat_views, log, ctx):
+    """Build each chart against the staging data. Returns the ones the file broke."""
+    failed = []
+    with connections["achilles"].cursor() as cursor:
+        for name in mat_views:
+            for definition in mat_views[name]:
+                try:
+                    cursor.execute(f"CREATE MATERIALIZED VIEW {name} AS {definition}")
+                    cursor.execute(f"DROP MATERIALIZED VIEW {name}")
+                except Exception as exc:
+                    error = translate(exc, f"verifying and uploading the file", log, view=name, **ctx)
+                    if not isinstance(error, TemporaryFailure):
+                        try:
+                            cursor.execute(f"DROP MATERIALIZED VIEW IF EXISTS {name}")
+                        except Exception:
+                            log.warning("Could not drop %s after failure", name)
+
+                    # A broken definition or a dead database stops everything,
+                    # a bad value is worth collecting so the user sees them all.
+                    if not isinstance(error, FileDataCorrupted):
+                        raise error from exc # stops immediately the processing
+                    failed.append((name, error.what_happened))
+                    break
+
+            if failed: # stop the processing at first error for now
+                return failed
+    return failed
+
+def _drop_staging_table(log):
+    try:
+        with connections["achilles"].cursor() as cursor:
+            cursor.execute("DROP TABLE IF EXISTS achilles_results_tmp CASCADE")
+            cursor.execute("DROP SEQUENCE IF EXISTS achilles_results_tmp_seq_id")
+    except Exception:
+        log.exception("Failed to drop the staging table")