Merge branch 'cookiecutterinstall' into 25-check-if-we-need-to-upgrade-to-python-313-and-validate-compatibility-pybis-custom-parsers-etc-if-compatibility-issues-arise-keep-python312-lucaszimm

Author: LucasZimm

.github/workflows/actions.yml

@@ -37,9 +37,11 @@ jobs:
         python -m mypy --ignore-missing-imports --follow-imports=silent --no-strict-optional openbis_upload_helper tests
     - name: Test with pytest
       run: |
-        python -m pytest -sv tests
+        python -m pip install pytest-django
+        python -m pytest -sv tests --reuse-db
     - name: Ruff linting
       run: |
+        pip install ruff
         ruff check .
     - name: Ruff formatting
       run: |

config/settings/base.py

@@ -326,9 +326,18 @@
 SOCIALACCOUNT_ADAPTER = "openbis_upload_helper.users.adapters.SocialAccountAdapter"
 # https://docs.allauth.org/en/latest/socialaccount/configuration.html
 SOCIALACCOUNT_FORMS = {
-    "signup": "openbis_upload_helper.users.forms.UserSocialSignupForm"
+    "signup": "openbis_upload_helper.users.forms.UserSocialSignupForm",
 }
 
 
 # Your stuff...
 # ------------------------------------------------------------------------------
+# [openbis]
+SECRET_ENCRYPTION_KEY = env(
+    "DJANGO_SECRET_KEY",
+    default="QZedkfcMcVvpwwrGieUCQdPmsHaz1ZwHwSSQZnKyq9sb4ZcKzjEEa4XRUnQ1ZYk0",
+)
+OPENBIS_URL = env(
+    "OPENBIS_URL",
+    default="https://openbis.example.org",
+)

config/settings/local.py

@@ -29,7 +29,8 @@
 # ------------------------------------------------------------------------------
 # https://docs.djangoproject.com/en/dev/ref/settings/#email-backend
 EMAIL_BACKEND = env(
-    "DJANGO_EMAIL_BACKEND", default="django.core.mail.backends.console.EmailBackend",
+    "DJANGO_EMAIL_BACKEND",
+    default="django.core.mail.backends.console.EmailBackend",
 )
 
 # WhiteNoise

openbis_upload_helper/app/admin.py

@@ -1,3 +1,3 @@
-from django.contrib import admin
+# for future use from django.contrib import admin
 
 # Register your models here.

openbis_upload_helper/app/models.py

@@ -1,3 +1,3 @@
-from django.db import models
+# for future use from django.db import models
 
 # Create your models here.

openbis_upload_helper/app/tests.py

@@ -1,3 +1,13 @@
 from django.test import TestCase
 
 # Create your tests here.
+
+
+def test_placeholder():
+    assert True
+
+
+class TestCaseClass(TestCase):
+    def test_example(self):
+        var_two = 2
+        assert var_two == 1 + 1

openbis_upload_helper/app/utils/__init__.py

@@ -1,12 +1,10 @@
-from .utils import (
-    FileLoader,
-    FileRemover,
-    FilesParser,
-    decrypt_password,
-    encrypt_password,
-    extract_name,
-    get_openbis_from_cache,
-    log_results,
-    preload_context_request,
-    reorganize_spaces,
-)
+from .utils import FileLoader as FileLoader
+from .utils import FileRemover as FileRemover
+from .utils import FilesParser as FilesParser
+from .utils import decrypt_password as decrypt_password
+from .utils import encrypt_password as encrypt_password
+from .utils import extract_name as extract_name
+from .utils import get_openbis_from_cache as get_openbis_from_cache
+from .utils import log_results as log_results
+from .utils import preload_context_request as preload_context_request
+from .utils import reorganize_spaces as reorganize_spaces

openbis_upload_helper/app/utils/utils.py

@@ -1,12 +1,12 @@
 import datetime
-import os
 import re
 import shutil
 import tarfile
 import tempfile
 import time
 import uuid
 import zipfile
+from pathlib import Path
 
 from bam_masterdata.logger import log_storage
 from bam_masterdata.logger import logger
@@ -36,10 +36,11 @@ def decrypt_password(encrypted_password):
         return decrypted_password.decode("utf-8")
     except InvalidToken as e:
         logger.error(f"Decryption failed: {e!s}")
-        raise InvalidToken("Decryption failed due to an invalid token.")
+        msg = "Decryption failed due to an invalid token."
+        raise InvalidToken(msg) from e
     except Exception as e:
         logger.error(f"Decryption error: {e!s}")
-        raise e
+        raise
 
 
 def get_openbis_from_cache(request):
@@ -96,17 +97,18 @@ def __init__(self, uploaded_files, selected_files):
 
     def load_files(self):
         if not self.uploaded_files:
-            raise ValueError("No files uploaded.")
+            msg = "No files uploaded."
+            raise ValueError(msg)
         # start countdown
         self.start_time = time.time()
 
         file_sizes = 0
         for uploaded_file in self.uploaded_files:
             file_sizes += uploaded_file.size
             if self.size_limit and file_sizes > int(float(self.size_limit)):
-                raise ValueError(
-                    f"Uploaded files exceed the size limit of {int(float(self.size_limit))} bytes.",
-                )
+                limit = int(float(self.size_limit))
+                msg = f"Uploaded files exceed limit of {limit} bytes."
+                raise ValueError(msg)
 
         for uploaded_file in self.uploaded_files:
             if uploaded_file.name.endswith(".zip"):
@@ -117,74 +119,79 @@ def load_files(self):
                 self._process_regular_file(uploaded_file)
 
         if not self.saved_file_names:
-            raise ValueError("No files were saved. Processing may have failed.")
+            msg = "No files were saved. Processing may have failed."
+            raise ValueError(msg)
         return self.saved_file_names
 
     def _process_zip(self, uploaded_file):
-        tmp_dir = tempfile.mkdtemp()
+        tmp_dir = Path(tempfile.mkdtemp())
         self.temp_dirs.append(tmp_dir)
-        zip_path = os.path.join(tmp_dir, uploaded_file.name)
-        with open(zip_path, "wb") as f:
+        zip_path = tmp_dir / uploaded_file.name
+        with Path.open(zip_path, "wb") as f:
             for chunk in uploaded_file.chunks():
                 f.write(chunk)
 
         with zipfile.ZipFile(zip_path, "r") as zip_ref:
             for zip_info in zip_ref.infolist():
                 if not zip_info.is_dir():
-                    target_path = os.path.join(tmp_dir, zip_info.filename)
-                    os.makedirs(os.path.dirname(target_path), exist_ok=True)
+                    target_path = tmp_dir / zip_info.filename
+                    target_path.parent.mkdir(parents=True, exist_ok=True)
                     # read zip member in chunks to allow timeout checks
                     with (
                         zip_ref.open(zip_info) as src,
-                        open(target_path, "wb") as out_file,
+                        Path.open(target_path, "wb") as out_file,
                     ):
                         while True:
                             chunk = src.read(8192)
                             if not chunk:
                                 break
                             out_file.write(chunk)
                     if zip_info.filename in self.selected_files:
-                        self.saved_file_names.append((zip_info.filename, target_path))
+                        self.saved_file_names.append(
+                            (zip_info.filename, str(target_path)),
+                        )
 
     def _process_tar(self, uploaded_file):
-        suffix = os.path.splitext(uploaded_file.name)[1]
+        suffix = Path(uploaded_file.name).suffix
         tmp_tar_path = None
         try:
             with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_tar:
                 for chunk in uploaded_file.chunks():
                     tmp_tar.write(chunk)
                 tmp_tar_path = tmp_tar.name
 
-            tmp_dir = tempfile.mkdtemp()
+            tmp_dir = Path(tempfile.mkdtemp())
             self.temp_dirs.append(tmp_dir)
             with tarfile.open(tmp_tar_path, "r:*") as tar_ref:
                 for member in tar_ref.getmembers():
                     if member.isfile():
                         if extracted_file := tar_ref.extractfile(member):
-                            target_path = os.path.join(tmp_dir, member.name)
-                            os.makedirs(os.path.dirname(target_path), exist_ok=True)
-                            with open(target_path, "wb") as out_file:
+                            target_path = tmp_dir / member.name
+                            target_path.parent.mkdir(parents=True, exist_ok=True)
+                            with Path.open(target_path, "wb") as out_file:
                                 while True:
                                     chunk = extracted_file.read(8192)
                                     if not chunk:
                                         break
                                     out_file.write(chunk)
                             if member.name in self.selected_files:
-                                self.saved_file_names.append((member.name, target_path))
+                                self.saved_file_names.append(
+                                    (member.name, str(target_path)),
+                                )
 
         finally:
-            if tmp_tar_path and os.path.exists(tmp_tar_path):
-                os.remove(tmp_tar_path)
+            if tmp_tar_path and Path(tmp_tar_path).exists():
+                Path.unlink(Path(tmp_tar_path))
 
     def _process_regular_file(self, uploaded_file):
-        tmp_dir = tempfile.mkdtemp()
+        tmp_dir = Path(tempfile.mkdtemp())
         self.temp_dirs.append(tmp_dir)
-        target_path = os.path.join(tmp_dir, uploaded_file.name)
-        with open(target_path, "wb") as f:
+        target_path = tmp_dir / uploaded_file.name
+        with Path.open(target_path, "wb") as f:
             for chunk in uploaded_file.chunks():
                 f.write(chunk)
         if uploaded_file.name in self.selected_files:
-            self.saved_file_names.append((uploaded_file.name, target_path))
+            self.saved_file_names.append((uploaded_file.name, str(target_path)))
 
     def antivirus_scan(self):
         # Placeholder for antivirus scanning logic
@@ -204,7 +211,8 @@ def assign_parsers(self, request):
         for idx, (file_name, file_path) in enumerate(self.uploaded_files):
             parser_name = request.POST.get(f"parser_type_{idx}")
             if not parser_name:
-                raise ValueError(f"No parser selected for file {file_name}")
+                msg = f"No parser selected for file {file_name}"
+                raise ValueError(msg)
 
             if parser_name not in self.parser_instances:
                 for parser in self.available_parsers.values():
@@ -225,20 +233,24 @@ def __init__(self, uploaded_files):
 
     def cleanup(self):
         for _, temp_file in self.uploaded_files:
-            temp_dir = os.path.dirname(temp_file)
-            if os.path.exists(temp_dir):
+            temp_dir = Path(temp_file).parent
+            if temp_dir.exists():
                 shutil.rmtree(temp_dir, ignore_errors=True)
         self.uploaded_files.clear()
 
 
-def log_results(request, parsed_files={}, context={}):
+def log_results(request, parsed_files=None, context=None):
+    if parsed_files is None:
+        parsed_files = {}
+    if context is None:
+        context = {}
     log_storage.clear()
     for parser, paths in parsed_files.items():
         for path in paths:
             log_storage.append(
                 {
-                    "event": f"[{parser}] Parsed: {os.path.basename(path)}",
-                    "timestamp": datetime.datetime.now().isoformat(),
+                    "event": f"[{parser}] Parsed: {Path(path).name}",
+                    "timestamp": datetime.datetime.now(tz=datetime.UTC).isoformat(),
                     "level": "info",
                 },
             )
@@ -299,8 +311,7 @@ def reorganize_spaces(spaces: list[str]) -> list[str]:
 
     # Sort by numeric value, not lexicographically (VP.2 < VP.10)
     def vp_sort_key(vp_key: str) -> int:
-        n = int(vp_key.split(".")[1])
-        return n
+        return int(vp_key.split(".")[1])
 
     vp_spaces: list[str] = []
     for vp_key in sorted(vp_groups.keys(), key=vp_sort_key):

openbis_upload_helper/app/views.py

@@ -1,4 +1,5 @@
 import uuid
+from typing import Any
 
 from bam_masterdata.cli.cli import run_parser
 from bam_masterdata.logger import logger
@@ -59,7 +60,7 @@ def login(request):
             cache.set(session_id, o, timeout=60 * 60)  # Cache for 1 hour (adjustable)
             return redirect("homepage")
 
-        except Exception as e:
+        except (ValueError, RuntimeError, ConnectionError) as e:
             logger.error(f"Login failed for user '{username}': {e}", exc_info=True)
             error = "Invalid username/password or personal access token."
 
@@ -72,13 +73,13 @@ def logout_view(request):
     return redirect("login")
 
 
-def homepage(request):
+def homepage(request):  # noqa: C901, PLR0911, PLR0912, PLR0915  (too complex, too many branches, too many statements)
     # Check if the user is logged in
     o = get_openbis_from_cache(request)
     if not o:
         logger.info("User not logged in, redirecting to login page.")
         return redirect("login")
-    context = {}
+    context: dict[str, Any] = {}
     available_parsers, parser_choices = preload_context_request(request, context)
 
     # load
@@ -114,18 +115,8 @@ def homepage(request):
                 collections_raw = o.get_experiments()
 
             # Filter
-            try:
-                projects = [extract_name(p) for p in projects_raw]
-            except Exception:
-                projects = [extract_name(p) for p in projects_raw]
-
-            try:
-                collections = [extract_name(c) for c in collections_raw]
-            except Exception:
-                collections = [extract_name(c) for c in collections_raw]
-
-            # context["projects"] = projects
-            # context["collections"] = collections
+            projects = [extract_name(p) for p in projects_raw]
+            collections = [extract_name(c) for c in collections_raw]
 
     else:
         projects = []
@@ -166,7 +157,7 @@ def homepage(request):
             request.session.pop("checker_logs", None)
             return redirect("homepage")
 
-        except Exception as e:
+        except (OSError, ValueError) as e:
             logger.exception("Error while uploading files")
             context["error"] = str(e)
             return render(request, "homepage.html", context)
@@ -204,7 +195,7 @@ def homepage(request):
             request.session["parsers_assigned"] = True
             return redirect("homepage")
 
-        except Exception as e:
+        except (ValueError, RuntimeError) as e:
             logger.exception("Error while assigning parsers")
             context["error"] = str(e)
             return render(request, "homepage.html", context)

openbis_upload_helper/uploader/entry_points/__init__.py

@@ -1 +1 @@
-from .load import get_entry_point_parsers
+from .load import get_entry_point_parsers as get_entry_point_parsers