Add CLI for launching nextflow jobs via seqera platform

This generally aligns more closely with seqera cli's `tw launch` than
with existing analysis-runner conventions.
https://docs.seqera.io/platform-cli/reference/launch

Though there is some analysis-runner logic applied - or at least
scaffolded to be applied in the server side implementation.

- Rather than specifying compute env and workspace, you specify dataset
  and access level, and the server will look up the compute env and
  workspace from those.
- Overriding of config with local config is disallowed for access levels
  other than test. This is because some config fields could be used to
  execute arbitrary code on the workflow runner. For access levels
  standard/full, a config can be specified but it must abide by the same
  rules as the workflow itself, ie. being on the main branch of an
  allow-listed repository.
- Several other `tw launch` args are omitted as we don't have a clear
  use case yet, they can be added later if needed.

Author: dancoates

packages/analysis-runner/pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     # a much newer version
     "grpcio-status>=1.48,<1.50",
     "hail>=0.2.134",
+    "pyyaml",
     "requests",
     "tabulate",
     "toml",

packages/analysis-runner/src/analysis_runner/cli.py

@@ -16,6 +16,7 @@
 )
 from analysis_runner.cli_config import add_config_args, run_config_from_args
 from analysis_runner.cli_cromwell import add_cromwell_args, run_cromwell_from_args
+from analysis_runner.cli_seqera import add_seqera_args, run_seqera_from_args
 
 
 def main_from_args(args: Sequence[str] | None = None):
@@ -37,6 +38,7 @@ def main_from_args(args: Sequence[str] | None = None):
     modes: dict[str, tuple[Callable[[], argparse.ArgumentParser], Callable]] = {
         'analysis-runner': (add_analysis_runner_args, run_analysis_runner_from_args),
         'cromwell': (add_cromwell_args, run_cromwell_from_args),
+        'seqera': (add_seqera_args, run_seqera_from_args),
         'config': (add_config_args, run_config_from_args),
     }
 

packages/analysis-runner/src/analysis_runner/cli_seqera.py

@@ -0,0 +1,199 @@
+"""
+CLI options for launching Nextflow workflows on the Seqera platform
+"""
+
+import argparse
+import sys
+from typing import Any
+
+import requests
+import yaml
+
+from analysis_runner.util import (
+    SERVER_ENDPOINT,
+    _perform_version_check,
+    confirm_choice,
+    get_server_endpoint,
+    logger,
+)
+from cpg_utils.cloud import get_google_identity_token
+
+
+def add_seqera_args(
+    parser: argparse.ArgumentParser | None = None,
+) -> argparse.ArgumentParser:
+    """
+    Add CLI arguments for launching Nextflow workflows on Seqera.
+
+    Flag names more or less mirror the Seqera Platform CLI (``tw launch``)
+    """
+    if not parser:
+        parser = argparse.ArgumentParser('seqera analysis-runner')
+
+    parser.add_argument(
+        '--dataset',
+        required=True,
+        type=str,
+        help='The dataset name, determines what data the run should have access to.',
+    )
+    parser.add_argument(
+        '--access-level',
+        choices=(['test', 'standard', 'full']),
+        default='test',
+        help='Which permissions level to grant when running the job.',
+    )
+
+    parser.add_argument(
+        '--repository',
+        '--repo',
+        required=True,
+        help='The name of the repository where the pipeline to run resides.',
+    )
+
+    parser.add_argument(
+        '--revision',
+        required=True,
+        help='The git branch or tag to use, for an exact commit use --commit-id.',
+    )
+    parser.add_argument(
+        '--commit-id',
+        required=True,
+        help='Specific Git commit hash to pin the pipeline execution to.',
+    )
+
+    parser.add_argument(
+        '--main-script',
+        required=False,
+        default='main.nf',
+        help='The Nextflow entry script to run. Defaults to "main.nf".',
+    )
+
+    parser.add_argument(
+        '--params-file',
+        required=False,
+        help='Path to a params file (YAML or JSON) forwarded to Nextflow as a '
+        '-params-file. Use "-" to read params from stdin.',
+    )
+
+    parser.add_argument(
+        '--config',
+        required=False,
+        help=(
+            'A full Nextflow config file to apply to the run, given as a github URL '
+            '(github.com/... or raw.githubusercontent.com/...). For standard / full '
+            'access the URL must point to a config on the main branch of an '
+            'allow-listed repository. Test access is less restricted: the URL may '
+            'reference any branch, or a local file path may be supplied instead.'
+            'Specifying a config will override pipeline config files.'
+        ),
+    )
+
+    parser.add_argument(
+        '--use-test-server',
+        action='store_true',
+        help='Use the test analysis-runner server',
+    )
+    parser.add_argument(
+        '--server-url',
+        required=False,
+        default=SERVER_ENDPOINT,
+        help='Supply a server URL to use, this will override the "--use-test-server"',
+    )
+
+    return parser
+
+
+def run_seqera_from_args(args: argparse.ArgumentParser):
+    """Run seqera nextflow submission from argparse.parse_arguments"""
+    return run_seqera(**vars(args))
+
+
+def _read_params(params: str) -> dict:
+    """Read a params file (YAML or JSON; "-" for stdin) into a dict."""
+    if params == '-':
+        content = sys.stdin.read()
+    else:
+        with open(params) as f:
+            content = f.read()
+
+    # JSON is a subset of YAML, so safe_load handles both formats.
+    parsed = yaml.safe_load(content)
+    if not isinstance(parsed, dict):
+        raise ValueError('The params file must contain a top-level mapping')
+    return parsed
+
+
+def run_seqera(
+    dataset: str,
+    access_level: str,
+    repository: str,
+    commit_id: str,
+    revision: str,
+    main_script: str = 'main.nf',
+    params_file: str | None = None,
+    config: str | None = None,
+    use_test_server: bool = False,
+    server_url: str | None = None,
+) -> None:
+    """
+    Prepare parameters and submit a Nextflow workflow to the analysis-runner.
+    """
+    _perform_version_check()
+
+    if access_level == 'full' and not confirm_choice(
+        'Full access increases the risk of accidental data loss. Continue?',
+    ):
+        raise SystemExit
+
+    server_args: dict[str, Any] = {
+        'dataset': dataset,
+        'access_level': access_level,
+        'main_script': main_script,
+        'repository': repository,
+        'commit_id': commit_id,
+        'revision': revision,
+    }
+
+    if params_file:
+        server_args['params'] = _read_params(params_file)
+
+    if config:
+        if config.startswith(('http://', 'https://')):
+            # A github URL, this is further validated on the server side to ensure
+            # the file is in an appropriate repo on an appropriate branch
+            server_args['config_url'] = config
+        elif access_level == 'test':
+            # Test access level can use a local config file
+            with open(config) as f:
+                server_args['config_text'] = f.read()
+        else:
+            raise SystemExit(
+                'For standard/full access, --config must be a github URL to a config '
+                'file on the main branch of an allow-listed repository.',
+            )
+
+    logger.info(
+        f'Submitting Nextflow workflow {repository}@{commit_id} on {revision} '
+        f'for dataset "{dataset}"',
+    )
+
+    server_endpoint = get_server_endpoint(
+        server_url=server_url, is_test=use_test_server
+    )
+    endpoint = server_endpoint.rstrip('/') + '/seqera'
+    _token = get_google_identity_token(server_endpoint)
+
+    response = requests.post(
+        endpoint,
+        json=server_args,
+        headers={'Authorization': f'Bearer {_token}'},
+        timeout=60,
+    )
+    try:
+        response.raise_for_status()
+        logger.info(f'Request submitted successfully: {response.text}')
+    except requests.HTTPError as e:
+        logger.critical(
+            f'Request failed with status {response.status_code}: {e!s}\n'
+            f'Full response: {response.text}',
+        )

packages/analysis-runner/tests/test_analysis_runner.py

@@ -1,15 +1,22 @@
 # ruff: noqa: S105
+import os
+import tempfile
 import unittest
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 from analysis_runner._version import __version__
 from analysis_runner.cli import main_from_args
 
 IMPORT_AR_IDENTITY_TOKEN_PATH = (
     'analysis_runner.cli_analysisrunner.get_google_identity_token'
 )
 IMPORT_CR_IDENTITY_TOKEN_PATH = 'analysis_runner.cli_cromwell.get_google_identity_token'
+IMPORT_SEQERA_IDENTITY_TOKEN_PATH = (
+    'analysis_runner.cli_seqera.get_google_identity_token'
+)
 
 REQUEST_POST_PATH = 'requests.post'
 REQUEST_GET_PATH = 'requests.get'
@@ -127,5 +134,115 @@ def test_submit_cli(self, mock_post: MagicMock, mock_identity_token: MagicMock):
         mock_identity_token.assert_called()
 
 
+class TestCliSeqera(unittest.TestCase):
+    SEQERA_ARGS = (
+        'seqera',
+        '--dataset',
+        'fewgenomes',
+        '--access-level',
+        'test',
+        '--repository',
+        'my-nf-pipeline',
+        '--commit-id',
+        'abc123',
+        '--revision',
+        'main',
+    )
+
+    @patch(IMPORT_SEQERA_IDENTITY_TOKEN_PATH)
+    @patch(REQUEST_POST_PATH)
+    def test_submit_cli(self, mock_post: MagicMock, mock_identity_token: MagicMock):
+        apply_mock_behaviour(
+            mock_post=mock_post,
+            mock_identity_token=mock_identity_token,
+        )
+
+        main_from_args(list(self.SEQERA_ARGS))
+
+        mock_post.assert_called()
+        mock_identity_token.assert_called()
+
+        # The request must hit the /seqera endpoint with the tower-style, snake_case
+        # arg set, and must NOT contain the standard analysis-runner / legacy keys.
+        _, kwargs = mock_post.call_args
+        assert mock_post.call_args[0][0].endswith('/seqera')
+        body = kwargs['json']
+        for key in (
+            'dataset',
+            'access_level',
+            'repository',
+            'commit_id',
+            'revision',
+            'main_script',
+        ):
+            assert key in body
+        for key in (
+            'cwd',
+            'cpu',
+            'memory',
+            'storage',
+            'output',
+            'output_dir',
+            'description',
+            'accessLevel',
+        ):
+            assert key not in body
+
+    @patch(IMPORT_SEQERA_IDENTITY_TOKEN_PATH)
+    @patch(REQUEST_POST_PATH)
+    def test_config_github_url_sent_as_config_url(
+        self,
+        mock_post: MagicMock,
+        mock_identity_token: MagicMock,
+    ):
+        apply_mock_behaviour(
+            mock_post=mock_post, mock_identity_token=mock_identity_token
+        )
+
+        url = 'https://github.com/org/configs/blob/main/conf/prod.config'
+        main_from_args([*self.SEQERA_ARGS, '--config', url])
+
+        body = mock_post.call_args.kwargs['json']
+        assert body['config_url'] == url
+        assert 'config_text' not in body
+
+    @patch(IMPORT_SEQERA_IDENTITY_TOKEN_PATH)
+    @patch(REQUEST_POST_PATH)
+    def test_config_local_file_inlined_for_test(
+        self,
+        mock_post: MagicMock,
+        mock_identity_token: MagicMock,
+    ):
+        apply_mock_behaviour(
+            mock_post=mock_post, mock_identity_token=mock_identity_token
+        )
+
+        with tempfile.NamedTemporaryFile('w', suffix='.config', delete=False) as f:
+            f.write('process {}')
+            config_path = f.name
+        self.addCleanup(os.unlink, config_path)
+
+        main_from_args([*self.SEQERA_ARGS, '--config', config_path])
+
+        body = mock_post.call_args.kwargs['json']
+        assert body['config_text'] == 'process {}'
+        assert 'config_url' not in body
+
+    @patch(IMPORT_SEQERA_IDENTITY_TOKEN_PATH)
+    @patch(REQUEST_POST_PATH)
+    def test_config_local_file_rejected_for_standard(
+        self,
+        mock_post: MagicMock,
+        mock_identity_token: MagicMock,
+    ):
+        apply_mock_behaviour(
+            mock_post=mock_post, mock_identity_token=mock_identity_token
+        )
+
+        args = [a if a != 'test' else 'standard' for a in self.SEQERA_ARGS]
+        with pytest.raises(SystemExit):
+            main_from_args([*args, '--config', '/some/local/file.config'])
+
+
 if __name__ == '__main__':
     unittest.main()