Sbp 481

.env.example

@@ -10,6 +10,10 @@ COMPUTE_ID=compute-env-id
 WORK_SPACE=workspace-id
 WORK_DIR=s3://bucket/workdir
 
+# NCI Gadi project code used for PBS submissions (clusterOptions) across all
+# Nextflow workflows. Optional; defaults to "yz52" if unset.
+GADI_PROJECT=yz52
+
 # AWS S3 Configuration for file uploads
 AWS_ACCESS_KEY_ID=your-access-key-id
 AWS_SECRET_ACCESS_KEY=your-secret-access-key

app/routes/workflows.py

@@ -14,7 +14,6 @@
 from pydantic import ValidationError
 from sqlalchemy import CursorResult, func, select, update
 from sqlalchemy.orm import Session
-from unidecode import unidecode
 
 from ..db.models import QueuedJob
 from ..db.models.core import AppUser, RunInput, RunMetric, S3Object, Workflow, WorkflowRun
@@ -234,28 +233,16 @@ async def launch_workflow(
         )
 
     user = db_session.execute(
-        select(AppUser.email, AppUser.name).where(AppUser.id == current_user_id)
+        select(AppUser.email).where(AppUser.id == current_user_id)
     ).one_or_none()
     if not user or not user.email:
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="Could not retrieve user details required for workflow launch.",
         )
-    user_email = user.email
-    # removes everything that isn't a letter, digit, or space
-    name = unidecode(user.name or "")
-    full_name = re.sub(r"[^a-zA-Z0-9 ]", "", name).replace(" ", "_")
-    institute = user_email.split("@")[-1] if "@" in user_email else None
-    ip_address: str | None = launch_ip or None
-
-    full_name = _require_launch_var("full_name", full_name)
-    institute = _require_launch_var("institute", institute)
-    ip_address = _require_launch_var("ip_address", ip_address)
     user_details = WorkflowUserDetails(
-        user_email=user_email,
-        full_name=full_name,
-        institute=institute,
-        ip_address=ip_address,
+        user_email=user.email,
+        ip_address=_require_launch_var("ip_address", launch_ip or None),
     )
 
     # Authoritative credit cost (server-side, non-spoofable). Only charged for
@@ -406,7 +393,7 @@ async def launch_workflow(
                     .values(
                         credit=AppUser.credit - run_credit_cost,
                         credit_updated_at=datetime.now(UTC),
-                        credit_updated_by=user_email,
+                        credit_updated_by=user_details.user_email,
                     )
                 ),
             )

app/run_scheduler.py

@@ -3,10 +3,17 @@
 import typer
 from apscheduler.triggers.cron import CronTrigger
 from apscheduler.triggers.interval import IntervalTrigger
+from dotenv import load_dotenv
 from loguru import logger
 
-from app.scheduler import SCHEDULER
-from app.scheduler.jobs import refresh_user_credits, submit_pending_jobs
+# Load .env before importing scheduler modules, which read required env vars (e.g.
+# SEQERA_API_URL) at call time. Unlike app/main.py, this entrypoint runs standalone
+# (`python -m app.run_scheduler`) and never imports main.py, so without this call
+# .env values are never loaded into the process environment.
+load_dotenv()
+
+from app.scheduler import SCHEDULER  # noqa: E402
+from app.scheduler.jobs import refresh_user_credits, submit_pending_jobs  # noqa: E402
 
 SUBMIT_INTERVAL = IntervalTrigger(minutes=5)
 MONTHLY_TRIGGER = CronTrigger(day=1, hour=1, minute=0, timezone="UTC")

app/schemas/workflows.py

@@ -97,8 +97,6 @@ class WorkflowUserDetails(BaseModel):
     """
 
     user_email: str = Field(..., description="Email address of the user")
-    full_name: str = Field(..., description="Full name of the user")
-    institute: str = Field(..., description="Institute of the user")
     ip_address: str = Field(..., description="IP address of the user")
 
 

app/services/bindflow_config.py

@@ -5,13 +5,14 @@
 from typing import Any
 
 from ..schemas.workflows import WorkflowUserDetails
+from .cluster_utils import GADI_PROJECT, encode_ip
 from .workflow_config_fetcher import fetch_workflow_config
 
 
 def get_bindflow_default_params(out_dir: str, samplesheet_url: str) -> dict[str, Any]:
     """Get default parameters for bindflow workflow."""
     return {
-        "project": "yz52",
+        "project": GADI_PROJECT,
         "outdir": out_dir,
         "input": samplesheet_url,
     }
@@ -25,17 +26,16 @@ def get_bindflow_config_profiles() -> list[str]:
 def get_bindflow_config_text(
     config_file_path: str,
     *,
-    job_id: str,
     user_details: WorkflowUserDetails,
-    timestamp: str,
 ) -> str:
     """Read bindflow base config and append a process override block with runtime values."""
     base = fetch_workflow_config(config_file_path)
 
-    cluster_opts = (
-        f"-v JOB_ID={job_id},USER_NAME={user_details.user_email},"
-        f"TIMESTAMP={timestamp},FULL_NAME={user_details.full_name},"
-        f"INSTITUTE={user_details.institute},IP_ADDRESS={user_details.ip_address}"
+    account = (
+        f"{user_details.user_email}:{encode_ip(user_details.ip_address)}"
+        if user_details.ip_address
+        else user_details.user_email
     )
+    cluster_opts = f"-A {account}"
     override = f'\nprocess {{\n    clusterOptions = "{cluster_opts}"\n}}\n'
     return base + override

app/services/bindflow_executor.py

@@ -58,14 +58,9 @@ async def prepare_bindflow_workflow(  # pylint: disable=too-many-locals
         raise SeqeraConfigurationError("Missing output identifier for workflow launch")
     out_dir = f"s3://{s3_bucket}/{output_key}"
 
-    timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
-
     dataset_url = f"s3://{s3_bucket}/{s3_input_key}"
     default_params = get_bindflow_default_params(out_dir, dataset_url)
 
-    default_params["job_id"] = run_name
-    default_params["user_name"] = user_details.user_email
-    default_params["timestamp"] = timestamp
     default_params["mode"] = mode
 
     # Merge any tool-specific params forwarded from the frontend form
@@ -91,9 +86,7 @@ async def prepare_bindflow_workflow(  # pylint: disable=too-many-locals
         "configProfiles": get_bindflow_config_profiles(),
         "configText": get_bindflow_config_text(
             config_path,
-            job_id=run_name,
             user_details=user_details,
-            timestamp=timestamp,
         ),
         "resume": False,
     }

app/services/cluster_utils.py

@@ -0,0 +1,13 @@
+"""Utilities for building HPC cluster option strings."""
+
+from __future__ import annotations
+
+import base64
+import os
+
+GADI_PROJECT: str = os.getenv("GADI_PROJECT", "yz52")
+
+
+def encode_ip(ip_address: str) -> str:
+    """Return the base64 encoding of an IP address string."""
+    return base64.b64encode(ip_address.encode()).decode()

app/services/proteinfold_config.py

@@ -5,14 +5,15 @@
 from typing import Any
 
 from ..schemas.workflows import WorkflowUserDetails
+from .cluster_utils import GADI_PROJECT, encode_ip
 from .workflow_config_fetcher import fetch_workflow_config
 
 
 def get_proteinfold_default_params(
     out_dir: str, samplesheet_url: str, mode: str = "alphafold2"
 ) -> dict[str, Any]:
     """Get default parameters for proteinfold workflow."""
-    return {"input": samplesheet_url, "outdir": out_dir, "project": "yz52", "mode": mode}
+    return {"input": samplesheet_url, "outdir": out_dir, "project": GADI_PROJECT, "mode": mode}
 
 
 def get_proteinfold_config_profiles() -> list[str]:
@@ -23,17 +24,16 @@ def get_proteinfold_config_profiles() -> list[str]:
 def get_proteinfold_config_text(
     config_file_path: str,
     *,
-    job_id: str,
     user_details: WorkflowUserDetails,
-    timestamp: str,
 ) -> str:
     """Read proteinfold base config and append a process override block with runtime values."""
     base = fetch_workflow_config(config_file_path)
 
-    cluster_opts = (
-        f"-P yz52 -v JOB_ID={job_id},USER_NAME={user_details.user_email},"
-        f"TIMESTAMP={timestamp},FULL_NAME={user_details.full_name},"
-        f"INSTITUTE={user_details.institute},IP_ADDRESS={user_details.ip_address}"
+    account = (
+        f"{user_details.user_email}:{encode_ip(user_details.ip_address)}"
+        if user_details.ip_address
+        else user_details.user_email
     )
+    cluster_opts = f"-P {GADI_PROJECT} -A {account}"
     override = f'\nprocess {{\n    clusterOptions = "{cluster_opts}"\n}}\n'
     return base + override

app/services/proteinfold_executor.py

@@ -50,14 +50,11 @@ def _build_params_text(
     mode: str,
     form_data: WorkflowFormData | None,
     custom_params: str | None,
-    extra_params: dict[str, Any] | None = None,
 ) -> str:
     """Build the YAML params string for the Seqera launch payload."""
     params = get_proteinfold_default_params(out_dir, samplesheet_url, mode)
     if form_data:
         params.update(_tool_params(form_data))
-    if extra_params:
-        params.update(extra_params)
     params_text = params_to_yaml_text(params)
     if custom_params and custom_params.strip():
         params_text = f"{params_text}\n{custom_params.rstrip()}"
@@ -89,9 +86,7 @@ async def prepare_proteinfold_workflow(
         raise SeqeraConfigurationError("Missing output identifier for workflow launch")
     out_dir = f"s3://{s3_bucket}/{output_id.strip()}"
 
-    timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
-    job_id = (form.runName or "").strip()
-    if not job_id:
+    if not form.runName or not form.runName.strip():
         raise SeqeraConfigurationError("Missing run name for workflow launch")
 
     sheet_url = f"s3://{s3_bucket}/{s3_input_key}"
@@ -101,11 +96,6 @@ async def prepare_proteinfold_workflow(
         mode,
         form_data,
         form.paramsText,
-        extra_params={
-            "job_id": job_id,
-            "user_name": user_details.user_email,
-            "timestamp": timestamp,
-        },
     )
 
     launch_payload: dict[str, Any] = {
@@ -119,9 +109,7 @@ async def prepare_proteinfold_workflow(
         "configProfiles": get_proteinfold_config_profiles(),
         "configText": get_proteinfold_config_text(
             config_path,
-            job_id=job_id,
             user_details=user_details,
-            timestamp=timestamp,
         ),
         "resume": False,
     }

app/services/wisps_config.py

@@ -5,6 +5,7 @@
 from typing import Any, Literal
 
 from ..schemas.workflows import WorkflowUserDetails
+from .cluster_utils import GADI_PROJECT, encode_ip
 from .workflow_config_fetcher import fetch_workflow_config
 
 WispsMode = Literal["g1-g2", "manual"]
@@ -39,9 +40,7 @@ def get_wisps_config_profiles() -> list[str]:
 def get_wisps_config_text(
     config_file_path: str,
     *,
-    job_id: str,
     user_details: WorkflowUserDetails,
-    timestamp: str,
 ) -> str:
     """Read wisps config and append a process override block with runtime values.
 
@@ -50,10 +49,11 @@ def get_wisps_config_text(
     """
     base = fetch_workflow_config(config_file_path)
 
-    cluster_opts = (
-        f"-P yz52 -v JOB_ID={job_id},USER_NAME={user_details.user_email},"
-        f"TIMESTAMP={timestamp},FULL_NAME={user_details.full_name},"
-        f"INSTITUTE={user_details.institute},IP_ADDRESS={user_details.ip_address}"
+    account = (
+        f"{user_details.user_email}:{encode_ip(user_details.ip_address)}"
+        if user_details.ip_address
+        else user_details.user_email
     )
+    cluster_opts = f"-P {GADI_PROJECT} -A {account}"
     override = f'\nprocess {{\n    clusterOptions = "{cluster_opts}"\n}}\n'
     return base + override

app/services/wisps_executor.py

@@ -54,7 +54,6 @@ async def prepare_wisps_workflow(
         raise SeqeraConfigurationError("Missing output identifier for workflow launch")
     out_dir = f"s3://{s3_bucket}/{output_id.strip()}"
 
-    timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
     job_id = (form.runName or "").strip()
     if not job_id:
         raise SeqeraConfigurationError("Missing run name for workflow launch")
@@ -72,9 +71,7 @@ async def prepare_wisps_workflow(
 
     config_text = get_wisps_config_text(
         config_path,
-        job_id=job_id,
         user_details=user_details,
-        timestamp=timestamp,
     )
 
     launch_payload: dict[str, Any] = {

pyproject.toml

@@ -22,7 +22,6 @@ dependencies = [
     "cachetools~=5.5",
     "python-jose[cryptography]~=3.5",
     "starlette-admin~=0.16",
-    "unidecode~=1.3",
     "apscheduler>=3.11.3",
     "loguru>=0.7.3",
     "typer>=0.26.8",

tests/test_proteinfold_coverage.py

@@ -35,8 +35,6 @@
 
 _USER_DETAILS = WorkflowUserDetails(
     user_email="user@ex.com",
-    full_name="Test_User",
-    institute="USYD",
     ip_address="1.2.3.4",
 )
 
@@ -369,7 +367,6 @@ async def test_prepare_proteinfold_workflow_writes_expected_queued_job(
             user_details=_USER_DETAILS.model_copy(
                 update={
                     "user_email": "test@example.com",
-                    "institute": "example.com",
                     "ip_address": "127.0.0.1",
                 }
             ),
@@ -544,33 +541,36 @@ def test_get_proteinfold_config_text_appends_process_block():
     with patch("builtins.open", mock_open(read_data="base_config")):
         result = get_proteinfold_config_text(
             "/fake/proteinfold.config",
-            job_id="my-job",
             user_details=_USER_DETAILS,
-            timestamp="20240101_120000",
         )
     assert "process {" in result
     assert "clusterOptions" in result
 
 
-def test_get_proteinfold_config_text_contains_job_fields():
+def test_get_proteinfold_config_text_contains_email_and_encoded_ip():
     with patch("builtins.open", mock_open(read_data="base_config")):
         result = get_proteinfold_config_text(
             "/fake/proteinfold.config",
-            job_id="my-job",
             user_details=_USER_DETAILS,
-            timestamp="20240101_120000",
         )
-    assert "my-job" in result
     assert "user@ex.com" in result
-    assert "20240101_120000" in result
+    assert "MS4yLjMuNA==" in result
+
+
+def test_get_proteinfold_config_text_without_ip_uses_email_only():
+    with patch("builtins.open", mock_open(read_data="base_config")):
+        result = get_proteinfold_config_text(
+            "/fake/proteinfold.config",
+            user_details=_USER_DETAILS.model_copy(update={"ip_address": ""}),
+        )
+    assert "-A user@ex.com" in result
+    assert ":" not in result.split("clusterOptions = ")[1]
 
 
 def test_get_proteinfold_config_text_contains_base_config():
     with patch("builtins.open", mock_open(read_data="base_config")):
         result = get_proteinfold_config_text(
             "/fake/proteinfold.config",
-            job_id="my-job",
             user_details=_USER_DETAILS,
-            timestamp="20240101_120000",
         )
     assert "base_config" in result