Replace 'rauth' with 'requests-oauthlib' for OAuth1 support and refactor related code for improved readability

Author: karlnyr

microSALT/utils/pubmlst/authentication.py

@@ -4,7 +4,7 @@
 from datetime import datetime, timedelta
 
 from dateutil import parser
-from rauth import OAuth1Session
+from requests_oauthlib import OAuth1Session
 
 from microSALT.config import Folders, PubMLSTCredentials, PasteurCredentials
 from microSALT.utils.pubmlst.constants import CREDENTIALS_KEY
@@ -27,7 +27,13 @@
 
 
 class ClientAuthentication:
-    def __init__(self, service: str, folders: Folders, pubmlst: PubMLSTCredentials, pasteur: PasteurCredentials):
+    def __init__(
+        self,
+        service: str,
+        folders: Folders,
+        pubmlst: PubMLSTCredentials,
+        pasteur: PasteurCredentials,
+    ):
         """Initialize the client with the specified service."""
         self.service: str = service
         self.folders = folders
@@ -53,10 +59,10 @@ def get_new_session_token(self, db: str):
             logger.debug(f"Requesting session token from URL: {url}")
 
             session = OAuth1Session(
-                consumer_key=consumer_key,
-                consumer_secret=consumer_secret,
-                access_token=access_token,
-                access_token_secret=access_secret,
+                client_key=consumer_key,
+                client_secret=consumer_secret,
+                resource_owner_key=access_token,
+                resource_owner_secret=access_secret,
             )
 
             response = session.get(url, headers={"User-Agent": "BIGSdb API downloader"})

microSALT/utils/pubmlst/client.py

@@ -2,7 +2,7 @@
 import logging
 import requests
 from werkzeug.exceptions import NotFound
-from rauth import OAuth1Session
+from requests_oauthlib import OAuth1Session
 
 from microSALT.config import Folders, PubMLSTCredentials, PasteurCredentials
 from microSALT.utils.pubmlst.authentication import ClientAuthentication
@@ -27,7 +27,14 @@
 class BaseClient:
     """Base client for interacting with authenticated APIs."""
 
-    def __init__(self, service: str, database: str = None, folders: Folders = None, pubmlst: PubMLSTCredentials = None, pasteur: PasteurCredentials = None):
+    def __init__(
+        self,
+        service: str,
+        database: str = None,
+        folders: Folders = None,
+        pubmlst: PubMLSTCredentials = None,
+        pasteur: PasteurCredentials = None,
+    ):
         """Initialize the client with the specified service."""
         try:
             self.service = service
@@ -101,10 +108,10 @@ def _make_request(
 
             # Create session with OAuth1
             session = OAuth1Session(
-                self.consumer_key,
-                self.consumer_secret,
-                access_token=token,
-                access_token_secret=secret,
+                client_key=self.consumer_key,
+                client_secret=self.consumer_secret,
+                resource_owner_key=token,
+                resource_owner_secret=secret,
             )
 
             response = session.request(method.value, url)
@@ -204,20 +211,39 @@ def list_schemes(self, db: str):
 class PubMLSTClient(BaseClient):
     """Client for interacting with the PubMLST authenticated API."""
 
-    def __init__(self, folders: Folders = None, pubmlst: PubMLSTCredentials = None, pasteur: PasteurCredentials = None):
+    def __init__(
+        self,
+        folders: Folders = None,
+        pubmlst: PubMLSTCredentials = None,
+        pasteur: PasteurCredentials = None,
+    ):
         """Initialize the PubMLST client."""
         super().__init__(service="pubmlst", folders=folders, pubmlst=pubmlst, pasteur=pasteur)
 
 
 class PasteurClient(BaseClient):
     """Client for interacting with the Pasteur authenticated API."""
 
-    def __init__(self, database: str, folders: Folders = None, pubmlst: PubMLSTCredentials = None, pasteur: PasteurCredentials = None):
+    def __init__(
+        self,
+        database: str,
+        folders: Folders = None,
+        pubmlst: PubMLSTCredentials = None,
+        pasteur: PasteurCredentials = None,
+    ):
         """Initialize the Pasteur client."""
-        super().__init__(service="pasteur", database=database, folders=folders, pubmlst=pubmlst, pasteur=pasteur)
+        super().__init__(
+            service="pasteur", database=database, folders=folders, pubmlst=pubmlst, pasteur=pasteur
+        )
 
 
-def get_client(service: str, database: str = None, folders: Folders = None, pubmlst: PubMLSTCredentials = None, pasteur: PasteurCredentials = None):
+def get_client(
+    service: str,
+    database: str = None,
+    folders: Folders = None,
+    pubmlst: PubMLSTCredentials = None,
+    pasteur: PasteurCredentials = None,
+):
     """Get the appropriate client for the specified service."""
     if service == "pasteur":
         return PasteurClient(database=database, folders=folders, pubmlst=pubmlst, pasteur=pasteur)

microSALT/utils/pubmlst/get_credentials.py

@@ -2,10 +2,15 @@
 import sys
 from argparse import ArgumentParser
 
-from rauth import OAuth1Service
-
-from microSALT.config import load_config
-from microSALT.config import Folders, PubMLSTCredentials, PasteurCredentials
+from requests_oauthlib import OAuth1Session
+
+from microSALT.config import (
+    Folders,
+    PasteurCredentials,
+    PubMLSTCredentials,
+    load_config,
+    MicroSALTConfig,
+)
 from microSALT.utils.pubmlst.constants import CREDENTIALS_KEY
 from microSALT.utils.pubmlst.helpers import get_path, get_service_config
 
@@ -18,48 +23,44 @@ def validate_credentials(client_id, client_secret):
         raise ValueError("Invalid CLIENT_SECRET: It must not be empty.")
 
 
-def get_request_token(service):
-    """Handle JSON response from the request token endpoint."""
-    response = service.get_raw_request_token(params={"oauth_callback": "oob"})
-    if not response.ok:
-        print(f"Error obtaining request token: {response.text}")
+def get_new_access_token(
+    client_id, client_secret, db: str, base_api: str, base_web: str
+) -> tuple[str, str]:
+    """Obtain a new access token and secret."""
+    # Step 1: fetch request token
+    oauth = OAuth1Session(client_id, client_secret=client_secret, callback_uri="oob")
+    response = oauth.fetch_request_token(f"{base_api}/db/{db}/oauth/get_request_token")
+    if not response:
+        print("Error obtaining request token.")
         sys.exit(1)
-    data = response.json()
-    return data["oauth_token"], data["oauth_token_secret"]
-
+    request_token = response["oauth_token"]
+    request_secret = response["oauth_token_secret"]
 
-def get_new_access_token(client_id, client_secret, db: str, base_api: str, base_web: str):
-    """Obtain a new access token and secret."""
-    service = OAuth1Service(
-        name="BIGSdb_downloader",
-        consumer_key=client_id,
-        consumer_secret=client_secret,
-        request_token_url=f"{base_api}/db/{db}/oauth/get_request_token",
-        access_token_url=f"{base_api}/db/{db}/oauth/get_access_token",
-        base_url=base_api,
-    )
-    request_token, request_secret = get_request_token(service)
     print(
         "Please log in using your user account at "
         f"{base_web}?db={db}&page=authorizeClient&oauth_token={request_token} "
         "using a web browser to obtain a verification code."
     )
     verifier = input("Please enter verification code: ")
 
-    raw_access = service.get_raw_access_token(
-        request_token, request_secret, params={"oauth_verifier": verifier}
+    # Step 2: exchange for access token
+    oauth = OAuth1Session(
+        client_id,
+        client_secret=client_secret,
+        resource_owner_key=request_token,
+        resource_owner_secret=request_secret,
+        verifier=verifier,
     )
-    if not raw_access.ok:
-        print(f"Error obtaining access token: {raw_access.text}")
+    access_data = oauth.fetch_access_token(f"{base_api}/db/{db}/oauth/get_access_token")
+    if not access_data:
+        print("Error obtaining access token.")
         sys.exit(1)
-
-    access_data = raw_access.json()
     return access_data["oauth_token"], access_data["oauth_token_secret"]
 
 
 def save_to_credentials_py(
     client_id, client_secret, access_token, access_secret, credentials_path, credentials_file
-):
+) -> None:
     """Save tokens in the credentials.py file."""
     credentials_path.mkdir(parents=True, exist_ok=True)
 
@@ -71,7 +72,7 @@ def save_to_credentials_py(
     print(f"Tokens saved to {credentials_file}")
 
 
-def main(service, config, species=None):
+def main(service: str, config: MicroSALTConfig, species: str | None = None):
     try:
         service_config = get_service_config(service, pubmlst=config.pubmlst, pasteur=config.pasteur)
         bigsd_config = service_config["config"]
@@ -89,7 +90,7 @@ def main(service, config, species=None):
         else:
             raise ValueError(f"Unknown service: {service}")
 
-        credentials_path = get_path(config.folders, CREDENTIALS_KEY)
+        credentials_path = get_path(folders=config.folders, config_key=CREDENTIALS_KEY)
         credentials_file = os.path.join(
             credentials_path, service_config.get("auth_credentials_file_name")
         )

microSALT/utils/pubmlst/helpers.py

@@ -18,10 +18,10 @@
 logger = logging.getLogger("main_logger")
 
 
-def get_path(config, config_key: str):
+def get_path(folders: Folders, config_key: str) -> Path:
     """Get and expand the file path from the configuration."""
     try:
-        path = getattr(config, config_key, None)
+        path = getattr(folders, config_key, None)
         if not path:
             raise PathResolutionError(config_key)
 
@@ -34,7 +34,7 @@ def get_path(config, config_key: str):
         raise PathResolutionError(config_key) from e
 
 
-def get_service_config(service: str, pubmlst: PubMLSTCredentials = None, pasteur: PasteurCredentials = None):
+def get_service_config(service: str, pubmlst: PubMLSTCredentials | None = None, pasteur: PasteurCredentials  | None= None):
     """
     Get the configuration for the specified service (e.g., 'pubmlst' or 'pasteur').
 

pyproject.toml

@@ -20,7 +20,7 @@ dependencies = [
     "pymysql>=1.0",
     "pyyaml>=6.0",
     "sqlalchemy>=2.0",
-    "rauth==0.7.3",
+    "requests-oauthlib>=1.3",
     "werkzeug>=3.1.6",
 ]