Brancheneinstufung2/connector-superoffice/health_check_so.py

import os
import requests
from dotenv import load_dotenv
import logging

# Set up basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class AuthHandler:
    def __init__(self):
        load_dotenv(override=True)
        self.client_id = os.getenv("SO_CLIENT_ID") or os.getenv("SO_SOD")
        self.client_secret = os.getenv("SO_CLIENT_SECRET")
        self.refresh_token = os.getenv("SO_REFRESH_TOKEN")
        self.redirect_uri = os.getenv("SO_REDIRECT_URI", "http://localhost")
        self.env = os.getenv("SO_ENVIRONMENT", "sod")
        self.cust_id = os.getenv("SO_CONTEXT_IDENTIFIER", "Cust55774")

        if not all([self.client_id, self.client_secret, self.refresh_token]):
            raise ValueError("SuperOffice credentials missing in .env file.")

        logger.info("AuthHandler initialized with environment variables.")

    def get_access_token(self):
        # This method would typically handle caching and refreshing
        # For this health check, we'll directly call _refresh_access_token
        return self._refresh_access_token()

    def _refresh_access_token(self):
        url = f"https://{self.env}.superoffice.com/login/common/oauth/tokens"
        data = {
            "grant_type": "refresh_token",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "refresh_token": self.refresh_token,
            "redirect_uri": self.redirect_uri
        }
        try:
            resp = requests.post(url, data=data)
            resp.raise_for_status()
            logger.info("Access token refreshed successfully.")
            return resp.json().get("access_token")
        except requests.exceptions.HTTPError as e:
            logger.error(f"❌ Token Refresh Error (Status: {e.response.status_code}): {e.response.text}")
            return None
        except Exception as e:
            logger.error(f"❌ Connection Error during token refresh: {e}")
            return None

class SuperOfficeClient:
    def __init__(self, auth_handler):
        self.auth_handler = auth_handler
        self.env = os.getenv("SO_ENVIRONMENT", "sod")
        self.cust_id = os.getenv("SO_CONTEXT_IDENTIFIER", "Cust55774")
        self.base_url = f"https://app-{self.env}.superoffice.com/{self.cust_id}/api/v1"
        self.access_token = self.auth_handler.get_access_token()
        if not self.access_token:
            raise Exception("Failed to obtain access token during SuperOfficeClient initialization.")
        
        self.headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        logger.info("✅ SuperOffice Client initialized and authenticated.")

    def _get(self, endpoint):
        try:
            resp = requests.get(f"{self.base_url}/{endpoint}", headers=self.headers)
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.HTTPError as e:
            logger.error(f"❌ API GET Error for {endpoint} (Status: {e.response.status_code}): {e.response.text}")
            return None
        except Exception as e:
            logger.error(f"❌ Connection Error for {endpoint}: {e}")
            return None

def perform_health_check():
    logger.info("Starting SuperOffice API health check...")
    try:
        auth_handler = AuthHandler()
        so_client = SuperOfficeClient(auth_handler)

        # Test 1: Associate/Me
        logger.info("\n--- Test 1: Fetching current user details (/Associate/Me) ---")
        user_details = so_client._get("Associate/Me")
        if user_details:
            logger.info(f"✅ Associate/Me successful! Connected as: {user_details.get('Name')} (Associate ID: {user_details.get('AssociateId')})")
        else:
            logger.error("❌ Associate/Me failed.")

        # Test 2: Get Person by ID (e.g., ID 1)
        logger.info("\n--- Test 2: Fetching Person with ID 1 (/Person/1) ---")
        person = so_client._get("Person/1")
        if person:
            logger.info(f"✅ Person/1 successful! Name: {person.get('Firstname')} {person.get('Lastname')}")
        else:
            logger.error("❌ Person/1 failed. (Could be that Person ID 1 does not exist or insufficient permissions)")

        # Test 3: Get Contact by ID (e.g., ID 1)
        logger.info("\n--- Test 3: Fetching Contact with ID 1 (/Contact/1) ---")
        contact = so_client._get("Contact/1")
        if contact:
            logger.info(f"✅ Contact/1 successful! Name: {contact.get('Name')}")
        else:
            logger.error("❌ Contact/1 failed. (Could be that Contact ID 1 does not exist or insufficient permissions)")
            
        # Overall check - if at least one read operation was successful
        if user_details or person or contact:
            logger.info("\n✅ SuperOffice API Connector seems partially operational (at least one read test passed).")
            return True
        else:
            logger.error("\n❌ SuperOffice API Connector is NOT operational (all read tests failed).")
            return False

    except ValueError as ve:
        logger.error(f"❌ Configuration error: {ve}")
        return False
    except Exception as e:
        logger.error(f"❌ An unexpected error occurred during health check: {e}", exc_info=True)
        return False

if __name__ == "__main__":
    if perform_health_check():
        logger.info("\nOverall SuperOffice API Health Check: PASSED (partially operational is still a pass for now).")
    else:
        logger.error("\nOverall SuperOffice API Health Check: FAILED.")