Copy script from gitlab project

.github/workflows/test.yml

@@ -33,10 +33,14 @@ jobs:
         run: |
           poetry run pytest
 
+      - name: Run import checker
+        run: |
+          poetry run isort --check .
+
       - name: Run code style checker
         run: |
-          poetry run flake8
+          poetry run pflake8
 
       - name: Run code formatter
         run: |
-          poetry run black --check src
+          poetry run black --check src tests

pyproject.toml

@@ -12,10 +12,12 @@ license = "GPL-3.0-or-later"
 
 [tool.poetry.dependencies]
 python = "^3.10"
+requests = "^2.31.0"
 
 [tool.poetry.dev-dependencies]
 black = "^22.3.0"
-flake8 = "^4.0.1"
+pyproject-flake8 = "^7.0.0"
+flake8-bugbear = "^24.4.0"
 pytest = "^7.1.1"
 isort = { version = "^5.10.1", extras = ["colors"] }
 
@@ -26,6 +28,14 @@ build-backend = "poetry.core.masonry.api"
 [tool.isort]
 profile = "black"
 
+[tool.black]
+line_length = 88
+
+[tool.flake8]
+max-line-length = 88
+extend-select = ["B950"]
+extend-ignore = ["E501"]
+
 [tool.pytest.ini_options]
 addopts = [
     "--import-mode=importlib",

src/npg_porch_cli/npg_porch_client.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+
+# Copyright (C) 2022, 2024 Genome Research Ltd.
+#
+# Author: Jennifer Liddle <[email protected]>
+#
+# npg_porch_client is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the Free
+# Software Foundation; either version 3 of the License, or (at your option) any
+# later version.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
+# details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program. If not, see <http://www.gnu.org/licenses/>.
+
+import argparse
+
+import requests
+
+# Certification file for https requests
+# if we don't have one, we can set certfile = False
+certfile = "/usr/share/ca-certificates/sanger.ac.uk/Genome_Research_Ltd_Certificate_Authority-cert.pem"  # noqa: B950
+
+# tokens from Porch
+admin_headers = {"Authorization": "Bearer a"}
+pipeline_headers = {"Authorization": "Bearer b"}
+
+# Command line arguments
+parser = argparse.ArgumentParser(description="Pipeline Orchestration Wrapper")
+parser.add_argument("--baseurl", type=str, help="base URL")
+parser.add_argument("--pipeline_url", type=str, help="Pipeline git project URL")
+parser.add_argument("--pipeline_version", type=str, help="Pipeline version")
+parser.add_argument("--pipeline", type=str, help="pipeline name")
+parser.add_argument("--idrun", type=int, help="id_run")
+parser.add_argument("--sample", type=str, help="sample name")
+parser.add_argument("--study", type=str, help="study ID")
+parser.add_argument(
+    "--status",
+    type=str,
+    help="new status to set",
+    default="DONE",
+    choices=["PENDING", "CLAIMED", "RUNNING", "DONE", "FAILED", "CANCELLED"],
+)
+parser.add_argument(
+    "command",
+    type=str,
+    help="command to send to npg_porch",
+    choices=["list", "plist", "register", "add", "claim", "update"],
+)
+args = parser.parse_args()
+
+if args.command == "list":
+    response = requests.get(
+        args.baseurl + "/tasks", verify=certfile, headers=pipeline_headers
+    )
+    if not response.ok:
+        print(f'"{response.reason}" received from {response.url}')
+        exit(1)
+
+    x = response.json()
+    for p in x:
+        if p["pipeline"]["name"] == args.pipeline:
+            print(f"{p['task_input']}\t{p['status']}")
+
+if args.command == "plist":
+    response = requests.get(
+        args.baseurl + "/pipelines", verify=certfile, headers=admin_headers
+    )
+    if not response.ok:
+        print(f'"{response.reason}" received from {response.url}')
+        exit(1)
+
+    x = response.json()
+    pipelines = {}
+    for p in x:
+        pname = p["name"]
+        if pname not in pipelines:
+            print(pname)
+            pipelines[pname] = 1
+
+if args.command == "register":
+    data = {
+        "name": args.pipeline,
+        "uri": args.pipeline_url,
+        "version": args.pipeline_version,
+    }
+    response = requests.post(
+        args.baseurl + "/pipelines", json=data, verify=certfile, headers=admin_headers
+    )
+    if not response.ok:
+        print(f'"{response.reason}" received from {response.url}')
+        exit(1)
+
+    print(response.json())
+
+if args.command == "add":
+    data = {
+        "pipeline": {
+            "name": args.pipeline,
+            "uri": args.pipeline_url,
+            "version": args.pipeline_version,
+        },
+        "task_input": {
+            "id_run": args.idrun,
+            "sample": args.sample,
+            "id_study": args.study,
+        },
+    }
+    response = requests.post(
+        args.baseurl + "/tasks", json=data, verify=certfile, headers=pipeline_headers
+    )
+    if not response.ok:
+        print(f'"{response.reason}" received from {response.url}')
+        exit(1)
+
+    print(response.json())
+
+if args.command == "claim":
+    data = {
+        "name": args.pipeline,
+        "uri": args.pipeline_url,
+        "version": args.pipeline_version,
+    }
+    response = requests.post(
+        args.baseurl + "/tasks/claim",
+        json=data,
+        verify=certfile,
+        headers=pipeline_headers,
+    )
+    if not response.ok:
+        print(f'"{response.reason}" received from {response.url}')
+        exit(1)
+
+    print(response.json())
+
+if args.command == "update":
+    data = {
+        "pipeline": {
+            "name": args.pipeline,
+            "uri": args.pipeline_url,
+            "version": args.pipeline_version,
+        },
+        "task_input": {
+            "id_run": args.idrun,
+            "sample": args.sample,
+            "id_study": args.study,
+        },
+        "status": args.status,
+    }
+    response = requests.put(
+        args.baseurl + "/tasks", json=data, verify=certfile, headers=pipeline_headers
+    )
+    if not response.ok:
+        print(f'"{response.reason}" received from {response.url}')
+        exit(1)
+
+    print(data)
+    print(response.json())

tests/test_api.py

@@ -1,5 +1,6 @@
 import npg_porch_cli.api as porchApi
 
+
 def test_addition():
 
     assert porchApi.add_two(1, 2) == 3