api_client.py

import requests
import base64
import json

class CROApiClient:
    """
    Client for interacting with the CRO (Companies Registration Office) API.
    """
    
    BASE_URL = "https://services.cro.ie/cws"
    
    def __init__(self, email, api_key):
        """
        Initialize the API client with credentials.
        
        Args:
            email (str): The email address for API authentication
            api_key (str): The API key for authentication
        """
        self.email = email
        self.api_key = api_key
        
        # Create the base64 encoded auth header
        auth_string = f"{email}:{api_key}"
        self.auth_header = base64.b64encode(auth_string.encode("utf-8")).decode("utf-8")
        
    def _get_headers(self):
        """
        Get the common headers for API requests.
        
        Returns:
            dict: Headers for API requests
        """
        return {
            "Authorization": f"Basic {self.auth_header}",
            "Accept": "application/json"
        }
        
    def _handle_response(self, response):
        """
        Handle the API response and check for errors.
        
        Args:
            response (Response): The response from the API
            
        Returns:
            dict: The JSON response data
            
        Raises:
            Exception: If the API returns an error
        """
        if response.status_code != 200:
            try:
                error_data = response.json()
                error_message = error_data.get("message", "Unknown error")
            except ValueError:
                error_message = response.text
                
            raise Exception(f"API Error ({response.status_code}): {error_message}")
        
        # Parse the JSON response
        data = response.json()
        
        # Handle different response formats
        if isinstance(data, list):
            # Convert list response to dict format for consistency
            return {"companies": data, "total_results": len(data)}
        
        return data
        
    def check_status(self):
        """
        Check if the CRO API services are running.
        
        Returns:
            bool: True if services are running, False otherwise
        """
        response = requests.get(
            f"{self.BASE_URL}/status",
            headers=self._get_headers()
        )
        
        data = self._handle_response(response)
        return data.get("ServicesRunning", False)
        
    def search_companies(self, company_name=None, company_num=None, company_bus_ind="e", 
                        search_type=3, address=None, skip=0, max_results=10, 
                        sort_by=None, sort_dir=None):
        """
        Search for companies using various parameters.
        
        Args:
            company_name (str, optional): Part of company/business name (min 3 characters)
            company_num (str, optional): Company or business number
            company_bus_ind (str, optional): Filter by type (C=companies, B=business names, E=both)
            search_type (int, optional): 1=Exact match, 2=Starts with, 3=Contains
            address (str, optional): Any part of address to narrow results
            skip (int, optional): Number of rows to skip (for pagination)
            max_results (int, optional): Maximum rows to return
            sort_by (str, optional): Field to sort by
            sort_dir (str, optional): ASC or DESC
            
        Returns:
            dict: Search results
        """
        params = {
            "format": "json"
        }
        
        # Add optional parameters if provided
        if company_name:
            params["company_name"] = company_name
        if company_num:
            params["company_num"] = company_num
        if company_bus_ind:
            params["company_bus_ind"] = company_bus_ind
        if search_type:
            params["searchType"] = search_type
        if address:
            params["address"] = address
        if skip:
            params["skip"] = skip
        if max_results:
            params["max"] = max_results
        if sort_by:
            params["sortBy"] = sort_by
        if sort_dir:
            params["sortDir"] = sort_dir
            
        response = requests.get(
            f"{self.BASE_URL}/companies",
            params=params,
            headers=self._get_headers()
        )
        
        return self._handle_response(response)
        
    def get_company_details(self, company_number, company_bus_ind):
        """
        Get detailed information about a specific company.
        
        Args:
            company_number (str): The company number
            company_bus_ind (str): The company business indicator (c or b)
            
        Returns:
            dict: Company details
        """
        response = requests.get(
            f"{self.BASE_URL}/company/{company_number}/{company_bus_ind}",
            params={"format": "json"},
            headers=self._get_headers()
        )
        
        return self._handle_response(response)