add decorators for imports and fix pylint errors

Author: marcelmamula

plugins/module_utils/auth.py

@@ -1,16 +1,54 @@
 import json
 import re
+import traceback
+from functools import wraps
 
 from urllib.parse import parse_qs, quote_plus, urljoin
-from bs4 import BeautifulSoup
-from requests.models import HTTPError
 
 from . import constants as C
 from . import exceptions
 
+try:
+    from bs4 import BeautifulSoup
+    HAS_BS4 = True
+except ImportError:
+    HAS_BS4 = False
+    BS4_IMPORT_ERROR = traceback.format_exc()
+    BeautifulSoup = None
+
+try:
+    from requests.models import HTTPError
+    HAS_REQUESTS = True
+except ImportError:
+    HAS_REQUESTS = False
+    REQUESTS_IMPORT_ERROR = traceback.format_exc()
+    HTTPError = None
+
 _GIGYA_SDK_BUILD_NUMBER = None
 
 
+def require_bs4(func):
+    # A decorator to check for the 'beautifulsoup4' library before executing a function.
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        if not HAS_BS4:
+            raise exceptions.SapLaunchpadError(f"The 'beautifulsoup4' library is required. Error: {BS4_IMPORT_ERROR}")
+        return func(*args, **kwargs)
+    return wrapper
+
+
+def require_requests(func):
+    # A decorator to check for the 'requests' library before executing a function.
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        if not HAS_REQUESTS:
+            raise exceptions.SapLaunchpadError(f"The 'requests' library is required. Error: {REQUESTS_IMPORT_ERROR}")
+        return func(*args, **kwargs)
+    return wrapper
+
+
+@require_requests
+@require_bs4
 def login(client, username, password):
     # Main authentication function.
     #
@@ -69,6 +107,8 @@ def login(client, username, password):
         client.post(endpoint, data=meta, headers=C.GIGYA_HEADERS)
 
 
+@require_requests
+@require_bs4
 def get_sso_endpoint_meta(client, url, **kwargs):
     # Scrapes an HTML page to find the next SSO form action URL and its input fields.
     method = 'POST' if kwargs.get('data') or kwargs.get('json') else 'GET'
@@ -100,6 +140,7 @@ def get_sso_endpoint_meta(client, url, **kwargs):
     return (endpoint, metadata)
 
 
+@require_requests
 def _get_gigya_login_params(client, url, data):
     # Follows a redirect and extracts parameters from the resulting URL's query string.
     gigya_idp_res = client.post(url, data=data)
@@ -109,9 +150,10 @@ def _get_gigya_login_params(client, url, data):
     return params
 
 
+@require_requests
 def _gigya_websdk_bootstrap(client, params):
     # Performs the initial bootstrap call to the Gigya WebSDK.
-    page_url = f'{C.URL_ACCOUNT_SAML_PROXY}?apiKey=' + params['apiKey'],
+    page_url = f'{C.URL_ACCOUNT_SAML_PROXY}?apiKey=' + params['apiKey']
     params.update({
         'pageURL': page_url,
         'sdk': 'js_latest',
@@ -124,6 +166,7 @@ def _gigya_websdk_bootstrap(client, params):
                headers=C.GIGYA_HEADERS)
 
 
+@require_requests
 def _gigya_login(client, username, password, api_key):
     # Performs a login using the standard Gigya accounts.login API.
     # This avoids a custom SAP endpoint that triggers password change notifications.
@@ -154,6 +197,7 @@ def _gigya_login(client, username, password, api_key):
     return login_response.get('login_token')
 
 
+@require_requests
 def _get_id_token(client, saml_params, login_token):
     # Exchanges a Gigya login token for a JWT ID token.
     query_params = {
@@ -166,6 +210,7 @@ def _get_id_token(client, saml_params, login_token):
     return token
 
 
+@require_requests
 def _get_uid(client, saml_params, login_token):
     # Retrieves the user's unique ID (UID) using the login token.
     query_params = {
@@ -177,6 +222,7 @@ def _get_uid(client, saml_params, login_token):
     return uid
 
 
+@require_requests
 def _get_uid_details(client, uid, id_token):
     # Fetches detailed account information for a given UID.
     url = f'{C.URL_ACCOUNT_CORE_API}/accounts/{uid}'
@@ -187,16 +233,18 @@ def _get_uid_details(client, uid, id_token):
     return uid_details_response
 
 
+@require_requests
 def _is_uid_linked_multiple_sids(uid_details):
     # Checks if a Universal ID (UID) is linked to more than one S-User ID.
     accounts = uid_details['accounts']
     linked = []
-    for _, v in accounts.items():
+    for _account_type, v in accounts.items():
         linked.extend(v['linkedAccounts'])
 
     return len(linked) > 1
 
 
+@require_requests
 def _select_account(client, uid, sid, id_token):
     # Selects a specific S-User ID when a Universal ID is linked to multiple accounts.
     url = f'{C.URL_ACCOUNT_CORE_API}/accounts/{uid}/selectedAccount'
@@ -207,6 +255,7 @@ def _select_account(client, uid, sid, id_token):
     return client.request('PUT', url, headers=headers, json=data)
 
 
+@require_requests
 def _get_sdk_build_number(client, api_key):
     # Fetches the gigya.js file to extract and cache the SDK build number.
     global _GIGYA_SDK_BUILD_NUMBER
@@ -224,6 +273,7 @@ def _get_sdk_build_number(client, api_key):
     return build_number
 
 
+@require_requests
 def _cdc_api_request(client, endpoint, saml_params, query_params):
     # Helper to make requests to the Gigya/CDC API, handling common parameters and errors.
     url = '/'.join((C.URL_ACCOUNT_CDC_API, endpoint))

plugins/module_utils/client.py

@@ -1,14 +1,35 @@
-import requests
 import re
-import urllib3
+import traceback
 
 from urllib.parse import urlparse
-from requests.adapters import HTTPAdapter
 
 from .constants import COMMON_HEADERS
-
-
-class _SessionAllowBasicAuthRedirects(requests.Session):
+from . import exceptions
+
+try:
+    import requests
+    from requests.adapters import HTTPAdapter
+    _RequestsSession = requests.Session
+    HAS_REQUESTS = True
+except ImportError:
+    HAS_REQUESTS = False
+    REQUESTS_IMPORT_ERROR = traceback.format_exc()
+    # Placeholders to prevent errors on module load
+    requests = None
+    HTTPAdapter = object
+    _RequestsSession = object
+
+try:
+    import urllib3
+    HAS_URLLIB3 = True
+except ImportError:
+    HAS_URLLIB3 = False
+    URLLIB3_IMPORT_ERROR = traceback.format_exc()
+    # Placeholder to prevent errors on module load
+    urllib3 = None
+
+
+class _SessionAllowBasicAuthRedirects(_RequestsSession):
     # By default, the `Authorization` header for Basic Auth will be removed
     # if the redirect is to a different host.
     # In our case, the DirectDownloadLink with `softwaredownloads.sap.com` domain
@@ -17,7 +38,8 @@ class _SessionAllowBasicAuthRedirects(requests.Session):
     # for sap.com domains.
     # This is only required for legacy API.
     def rebuild_auth(self, prepared_request, response):
-        if 'Authorization' in prepared_request.headers:
+        # The parent class might not be a real requests.Session if requests is not installed.
+        if HAS_REQUESTS and 'Authorization' in prepared_request.headers:
             request_hostname = urlparse(prepared_request.url).hostname
             if not re.match(r'.*sap.com$', request_hostname):
                 del prepared_request.headers['Authorization']
@@ -28,6 +50,9 @@ def _is_updated_urllib3():
     # and will be removed in v2.0.0.
     # Typically, the default version on RedHat 8.2 is 1.24.2,
     # so we need to check the version of urllib3 to see if it's updated.
+    if not HAS_URLLIB3:
+        return False
+
     urllib3_version = urllib3.__version__.split('.')
     if len(urllib3_version) == 2:
         urllib3_version.append('0')
@@ -44,6 +69,11 @@ class ApiClient:
     # object-oriented interface for making API requests, replacing the
     # previous global session and request functions.
     def __init__(self):
+        if not HAS_REQUESTS:
+            raise exceptions.SapLaunchpadError(f"The 'requests' library is required. Error: {REQUESTS_IMPORT_ERROR}")
+        if not HAS_URLLIB3:
+            raise exceptions.SapLaunchpadError(f"The 'urllib3' library is required. Error: {URLLIB3_IMPORT_ERROR}")
+
         self.session = _SessionAllowBasicAuthRedirects()
 
         # Configure retry logic for the session.

plugins/module_utils/maintenance_planner/api.py

@@ -1,20 +1,54 @@
 import re
 import time
+import traceback
 from html import unescape
+from functools import wraps
 from urllib.parse import urljoin
-from bs4 import BeautifulSoup
-from lxml import etree
 
 from .. import constants as C
 from .. import exceptions
 from ..auth import get_sso_endpoint_meta
 
+try:
+    from bs4 import BeautifulSoup
+    HAS_BS4 = True
+except ImportError:
+    HAS_BS4 = False
+    BS4_IMPORT_ERROR = traceback.format_exc()
+
+try:
+    from lxml import etree
+    HAS_LXML = True
+except ImportError:
+    HAS_LXML = False
+    LXML_IMPORT_ERROR = traceback.format_exc()
+
 # Module-level cache
 _MP_XSRF_TOKEN = None
 _MP_TRANSACTIONS = None
 _MP_NAMESPACE = 'http://xml.sap.com/2012/01/mnp'
 
 
+def require_bs4(func):
+    # A decorator to check for the 'beautifulsoup4' library before executing a function.
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        if not HAS_BS4:
+            raise exceptions.SapLaunchpadError(f"The 'beautifulsoup4' library is required. Error: {BS4_IMPORT_ERROR}")
+        return func(*args, **kwargs)
+    return wrapper
+
+
+def require_lxml(func):
+    # A decorator to check for the 'lxml' library before executing a function.
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        if not HAS_LXML:
+            raise exceptions.SapLaunchpadError(f"The 'lxml' library is required. Error: {LXML_IMPORT_ERROR}")
+        return func(*args, **kwargs)
+    return wrapper
+
+
 def auth_userapps(client):
     # Authenticates against userapps.support.sap.com to establish a session.
     _clear_mp_cookies(client, 'userapps')
@@ -32,6 +66,7 @@ def auth_userapps(client):
     client.post(endpoint, data=meta)
 
 
+@require_bs4
 def get_transactions(client):
     # Retrieves a list of all available Maintenance Planner transactions.
     global _MP_TRANSACTIONS
@@ -67,6 +102,7 @@ def get_transaction_id(client, name):
     raise exceptions.FileNotFoundError(f"Transaction '{name}' not found by name or display ID.")
 
 
+@require_lxml
 def get_transaction_filename_url(client, trans_id):
     # Parses the files XML to get a list of (URL, Filename) tuples.
     xml = _get_download_files_xml(client, trans_id)
@@ -175,6 +211,7 @@ def _get_transaction(client, key, value):
     raise exceptions.FileNotFoundError(f"Transaction with {key}='{value}' not found.")
 
 
+@require_lxml
 def _build_mnp_xml(**params):
     # Constructs the MNP XML payload for API requests.
     mnp = f'{{{_MP_NAMESPACE}}}'

plugins/module_utils/maintenance_planner/main.py

@@ -3,8 +3,13 @@
 from .. import auth, exceptions
 from ..client import ApiClient
 from . import api
-from requests.exceptions import HTTPError
 
+try:
+    from requests.exceptions import HTTPError
+    HAS_REQUESTS = True
+except ImportError:
+    HAS_REQUESTS = False
+    HTTPError = None
 
 def run_files(params):
     # Runner for maintenance_planner_files module.
@@ -14,6 +19,11 @@ def run_files(params):
         msg=''
     )
 
+    if not HAS_REQUESTS:
+        result['failed'] = True
+        result['msg'] = "The 'requests' library is required for this module."
+        return result
+
     client = ApiClient()
     username = params['suser_id']
     password = params['suser_password']
@@ -56,6 +66,11 @@ def run_stack_xml_download(params):
         msg=''
     )
 
+    if not HAS_REQUESTS:
+        result['failed'] = True
+        result['msg'] = "The 'requests' library is required for this module."
+        return result
+
     client = ApiClient()
     username = params['suser_id']
     password = params['suser_password']