Launch Seqera workflows from analysis-runner server

packages/server/main.py

@@ -9,6 +9,7 @@
 from ar import add_analysis_runner_routes
 from config import add_config_routes
 from cromwell import add_cromwell_routes
+from seqera import add_seqera_routes
 
 # Patching asyncio *before* importing the Hail Batch module is necessary to avoid a
 # "Cannot enter into task" error.
@@ -97,6 +98,7 @@ async def init_func():
     add_analysis_runner_routes(routes)
     add_cromwell_routes(routes)
     add_config_routes(routes)
+    add_seqera_routes(routes)
     app.add_routes(routes)
 
     return app

packages/server/seqera.py

@@ -0,0 +1,143 @@
+import asyncio
+
+import requests
+from aiohttp import web
+from cachetools.func import lru_cache
+from cloudpathlib import AnyPath
+from util import get_seqera_config, read_ar_secret
+
+
+class SeqeraApiClient:
+    def __init__(self, dataset: str, access_level: str):
+        seqera_config = get_seqera_config()
+        self.org_id = seqera_config['org_id']
+        self.api_url = seqera_config['api_url']
+        self.dataset_config = seqera_config['datasets'][dataset][access_level]
+
+        self.token = read_ar_secret(self.dataset_config['launch_token_secret_name'])
+
+    def get(self, endpoint: str, params: dict | None = None) -> dict:
+        headers = {
+            'Authorization': f'Bearer {self.token}',
+            'Accept': 'application/json',
+        }
+        url = f'{self.api_url}/{endpoint}'
+        response = requests.get(url, headers=headers, params=params)
+
+        response.raise_for_status()
+        return response.json()
+
+    def post(self, endpoint: str, body: dict, params: dict | None = None) -> dict:
+        headers = {
+            'Authorization': f'Bearer {self.token}',
+            'Content-Type': 'application/json',
+        }
+        url = f'{self.api_url}/{endpoint}'
+        response = requests.post(url, headers=headers, params=params, json=body)
+
+        if not response.ok and response.text:
+            reason = f'{response.status_code} {response.reason}: {response.text}'
+            raise web.HTTPBadRequest(reason=reason)
+        response.raise_for_status()
+        return response.json()
+
+    @property
+    def server_url(self) -> str:
+        return self.api_url.replace('://api.', '://', 1).removesuffix('/api')
+
+    @property
+    @lru_cache
+    def org_name(self) -> str:
+        return self.get(f'orgs/{self.org_id}')['organization']['name']
+
+    @property
+    @lru_cache
+    def workspace_name(self) -> str:
+        workspace_id = self.dataset_config['workspace_id']
+        response = self.get(f'orgs/{self.org_id}/workspaces/{workspace_id}')
+        return response['workspace']['name']
+
+    @property
+    def workspace_param(self) -> dict:
+        return {'workspaceId': self.dataset_config['workspace_id']}
+
+    def compute_environment(self, cenv_id: int) -> dict:
+        response = self.get(f'compute-envs/{cenv_id}', params=self.workspace_param)
+        return response['computeEnv']
+
+    SEQERA_KEYS = {
+        'commit_id': 'commitId',
+        'config_text': 'configText',
+        'main_script': 'mainScript',
+        'params': 'paramsText',
+        'repository': 'pipeline',
+        'revision': 'revision',
+    }
+
+    def launch_workflow(self, params: dict) -> str:
+        compute_env = self.compute_environment(self.dataset_config['compute_env_id'])
+
+        launch = {
+            'launch': {
+                'computeEnvId': compute_env['id'],
+                'workDir': compute_env['config']['workDir'],
+            }
+        }
+
+        for ar_key, seqera_key in self.SEQERA_KEYS.items():
+            if  ar_key in params:
+                launch['launch'][seqera_key] = params[ar_key]
+
+        return self.post('workflow/launch', launch, self.workspace_param)['workflowId']
+
+
+def add_seqera_routes(routes: web.RouteTableDef):
+    pass  # NUKEME
+
+if True:  # NUKEME
+    """Add Seqera route to 'routes' flask API."""
+
+    #@routes.post('/seqera')
+    async def seqera(request: web.Request) -> web.Response:
+        """Main seqera submission entry point."""
+
+        params = await request.json()
+
+        if 'config_url' in params:
+            # TODO Check repo/branch/etc permissions
+            params['config_text'] = AnyPath(params['config_url']).read_text()
+
+        seqera = SeqeraApiClient(params['dataset'], params['access_level'])
+        workflow_id = seqera.launch_workflow(params)
+
+        try:
+            where = f' at [{seqera.org_name} / {seqera.workspace_name}] workspace'
+            url = f'{seqera.server_url}/orgs/{seqera.org_name}/workspaces/{seqera.workspace_name}/watch/{workflow_id}'
+        except (requests.HTTPError, KeyError):
+            where = ''
+            url = '[URL unavailable]'
+
+        return web.Response(text=f'Workflow {workflow_id} submitted{where}.\n{url}\n')
+
+
+# For easy testing
+if __name__ == '__main__':
+
+    class BananaRequest(web.Request):
+        def __init__(self):
+            pass
+
+        async def json(self):
+            return {
+                'dataset': 'fewgenomes',
+                'access_level': 'test',
+                # parameter names below here TBD
+                #'repository': 'https://github.com/jmarshall/test',
+                #'commit_id': '21cd3b269ef07aab1f44e6e97755c2e52efac9c7',
+                'repository': 'https://github.com/nextflow-io/hello',
+                'commit_id': '3c2cdc9823c2b4636e5e3e73e223878099ff5dc9',
+            }
+
+    br = BananaRequest()
+    resp = asyncio.run(seqera(br))
+    print(resp.text)

packages/server/util.py

@@ -75,6 +75,30 @@ def get_server_config() -> dict:
     raise web.HTTPInternalServerError(reason='Failed to read server-config secret')
 
 
+# cache the result for 60 seconds, so we can call this function multiple times
+@ttl_cache(maxsize=1, ttl=600)
+def get_seqera_config() -> dict:
+    config = os.getenv('SEQERA_PLATFORM_CONFIG')
+    if config is None:
+        config = read_secret(ANALYSIS_RUNNER_PROJECT_ID, 'seqera-platform-config')
+    if config is None:
+        raise web.HTTPInternalServerError(
+            reason='Failed to read seqera-platform-config secret'
+        )
+
+    return json.loads(config)
+
+
+def read_ar_secret(secret_name: str) -> str:
+    """Read a secret stored in the analysis-runner project."""
+    value = read_secret(ANALYSIS_RUNNER_PROJECT_ID, secret_name)
+    if value is None:
+        raise web.HTTPInternalServerError(
+            reason='Failed to read analysis-runner secret'
+        )
+    return value
+
+
 async def _get_hail_version(environment: str) -> str:
     """ASYNC get hail version for the hail server in the local deploy_config"""
     if not environment == 'gcp':

pyproject.toml

@@ -37,6 +37,9 @@ ignore = [
 [tool.ruff.lint.pylint]
 max-positional-args = 8
 
+[tool.ruff.lint.flake8-annotations]
+mypy-init-return = true
+
 [tool.ruff.lint.flake8-quotes]
 inline-quotes = "single"