My revised version:

- Refactored the cli wrapper
- Refactored the settings to be in line with the rest
- Modified the unit tests

config/local-tests.yml

@@ -14,4 +14,7 @@ emg:
     results_path: 'fixtures/'
     celery_broker: 'redis://localhost:6379/0'
     celery_backend: 'redis://localhost:6379/1'
-  results_production_dir: '/dummy/path/results'
+  results_production_dir: '/dummy/path/results'
+  # metagenomics exchange
+  me_api: 'http://wp-np2-5c.ebi.ac.uk:8080/ena/registry/metagenome/api'
+  me_api_token: 'mgx 3D70473ED7C443CA9E97749F62FFCC5D'

emgapi/management/commands/mgx_api.py

@@ -101,6 +101,7 @@ def handle(self, *args, **options):
         self.assembly_accession = options.get("assembly")
         self.run_accession = options.get("run")
 
+        # FIXME: this command needs adjustments
         broker = options.get("broker")
         url = settings.ME_API['dev']
         check_url = url + f'brokers/{self.broker}/datasets'

emgapi/metagenomics_exchange.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-# Copyright 2018-2021 EMBL - European Bioinformatics Institute
+# Copyright 2018-2023 EMBL - European Bioinformatics Institute
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,98 +14,68 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import sys
 import logging
 import requests
 
+from django.conf import settings
+
+
 class MetagenomicsExchangeAPI:
+
     """Metagenomics Exchange API Client"""
-    def get_request(self, session, url, params):
-        response = session.get(url, params=params)
-        data = None
-        if not response.ok:
-            logging.error(
-                "Error retrieving dataset {}, response code: {}".format(
-                    url, response.status_code
-                )
-            )
-        else:
-            data = response.json()
-        return data
 
-    def post_request(self, session, url, data):
-        default = {
-            "headers": {
-                "Content-Type": "application/json",
-                "Accept": "application/json",
-            }
-        }
+    def __init__(self, broker="EMG"):
+        self.base_url = settings.ME_API
+        self.__token = settings.ME_API_TOKEN
+        self.broker = broker
 
-        response = session.post(
-            url, json=data, **default
+    def get_request(self, endpoint: str, params: dict):
+        """Make a GET request, returns the response"""
+        headers = {"Accept": "application/json", "Authorization": self.__token}
+        response = requests.get(
+            f"{self.base_url}/{endpoint}", headers=headers, params=params
         )
+        response.raise_for_status()
+        return response
 
-        if response.ok:
-            print('Added')
-            logging.info("Data added to ME")
-            logging.debug(response.text)
-        else:
-            print(response.text)
+    def post_request(self, endpoint: str, data: dict):
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "Authorization": self.__token,
+        }
+        response = requests.post(
+            f"{self.base_url}/{endpoint}", json=data, headers=headers
+        )
+        response.raise_for_status()
         return response
-    def add_record(
-        self, url, mgya, run_accession, public, broker, token
-    ):
+
+    def add_record(self, mgya: str, run_accession: str, public: bool):
         data = {
-            'confidence': 'full',
-            'endPoint': f'https://www.ebi.ac.uk/metagenomics/analyses/mgya',
-            'method': ['other_metadata'],
-            'sourceID': mgya,
-            'sequenceID': run_accession,
-            'status': 'public' if public else 'private',
-            'brokerID': broker,
+            "confidence": "full",
+            "endPoint": f"https://www.ebi.ac.uk/metagenomics/analyses/mgya",
+            "method": ["other_metadata"],
+            "sourceID": mgya,
+            "sequenceID": run_accession,
+            "status": "public" if public else "private",
+            "brokerID": self.broker,
         }
+        response = self.post_request(endpoint="datasets", data=data)
+        return response.json()
 
-        with requests.Session() as session:
-            url = url + 'datasets'
-            session = self._authenticate_session(session, url, token)
-            print(session)
-            response = self.post_request(session=session, url=url, data=data)
-            data = response.json()
-            return data
-
-    def check_analysis(self, url, sourceID, public, token):
-        logging.info(f'Check {sourceID}')
+    def check_analysis(self, source_id: str, public: bool) -> bool:
+        logging.info(f"Check {source_id}")
         params = {
-            'status': 'public' if public else 'private',
+            "status": "public" if public else "private",
         }
-        with requests.Session() as session:
-            session = self._authenticate_session(session, url, token)
-            response = self.get_request(session=session, url=url, params=params)
-            if response:
-                data = response['datasets']
-                exists = False
-                for item in data:
-                    if item['sourceID'] == sourceID:
-                        exists = True
-                logging.info(f"{sourceID} exists: {exists}")
-        return exists
-
-    def _authenticate_session(self, session, url, token):
-        """Authenticate the MGnify API request"""
-
-        logging.debug(f"Authenticating ME account")
-
-        headers = {"Authorization": token}
-
-        response = session.get(url, headers=headers)
-
+        endpoint = f"brokers/{self.broker}/datasets"
+        response = self.get_request(endpoint=endpoint, params=params)
         if response.ok:
-            logging.debug("ME account successfully authenticated.")
-            print(f'Auth {url} {response}')
-        else:
-            print(response)
-            # Log textual reason of responded HTTP Status
-            logging.error(f"Authentication services responded: {response.reason}). Program will exit now.")
-            sys.exit(1)
+            data = response.json()
+            datasets = data.get("datasets")
+            for item in datasets:
+                if item.get("sourceID") == source_id:
+                    return True
+            logging.info(f"{source_id} exists")
 
-        return session
+        return False

emgcli/settings.py

@@ -643,8 +643,10 @@ def create_secret_key(var_dir):
     os.environ['ENA_API_PASSWORD'] = EMG_CONF['emg']['ena_api_password']
 
 # Metagenomics Exchange
-ME_API = {
-    'real': 'https://www.ebi.ac.uk/ena/registry/metagenome/api/',
-    'dev': 'http://wp-np2-5c.ebi.ac.uk:8080/ena/registry/metagenome/api/'
-}
-ME_TOKEN = 'mgx 3D70473ED7C443CA9E97749F62FFCC5D'
+try:
+    ME_API = EMG_CONF['emg']['me_api']
+    ME_API_TOKEN = EMG_CONF['emg']['me_api_token']
+except KeyError:
+    ME_API = ""
+    ME_API_TOKEN = ""
+    warnings.warn("The metagenomics exchange API and Token are not configured properly")

tests/me/test_metagenomics_exchange.py

@@ -1,30 +1,28 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-from emgapi import metagenomics_exchange as ME
-from django.conf import settings
+import pytest
+from unittest import mock
 
-class TestME:
+from emgapi.metagenomics_exchange import MetagenomicsExchangeAPI
+
+from requests import HTTPError
 
-    test_ME = ME.MetagenomicsExchangeAPI()
+
+class TestME:
     def test_check_existing_analysis(self):
-        sourceID = "MGYA00293719"
-        broker = 'EMG'
-        url = settings.ME_API['dev'] + f'brokers/{broker}/datasets'
-        token = settings.ME_TOKEN
-        assert self.test_ME.check_analysis(url, sourceID, True, token)
+        me_api = MetagenomicsExchangeAPI()
+        source_id = "MGYA00293719"
+        assert me_api.check_analysis(source_id, True)
 
     def test_check_not_existing_analysis(self):
-        sourceID = "MGYA00293719"
-        broker = 'MAR'
-        url = settings.ME_API['dev'] + f'brokers/{broker}/datasets'
-        token = settings.ME_TOKEN
-        assert not self.test_ME.check_analysis(url, sourceID, True, token)
+        me_api = MetagenomicsExchangeAPI(broker="MAR")
+        source_id = "MGYA00293719"
+        assert not me_api.check_analysis(source_id, True)
 
     def test_post_existing_analysis(self):
-        sourceID = "MGYA00293719"
-        broker = 'EMG'
-        url = settings.ME_API['dev']
-        token = settings.ME_TOKEN
-        assert not self.test_ME.add_record(url=url, mgya=sourceID, run_accession="ERR3063408", public=True,
-                                           broker=broker, token=token)
+        me_api = MetagenomicsExchangeAPI()
+        source_id = "MGYA00293719"
+        # Should return -> https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/409
+        with pytest.raises(HTTPError, match="409 Client Error"):
+            me_api.add_record(mgya=source_id, run_accession="ERR3063408", public=True)