import os import requests import logging from typing import Dict, Any, Optional from dotenv import load_dotenv # Load environment variables load_dotenv() # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class JobAPIService: """Service for calling third-party job APIs with basic authentication""" def __init__(self): # Get configuration from environment variables self.base_url = os.getenv("THIRD_PARTY_JOBS_URL") self.username = os.getenv("THIRD_PARTY_API_AUTH_USER_NAME") self.password = os.getenv("THIRD_PARTY_API_AUTH_PASSWORD") # Validate configuration if not self.base_url: raise ValueError("THIRD_PARTY_JOBS_URL not configured in environment variables") if not self.username or not self.password: raise ValueError("Third-party API credentials not configured in environment variables") # Clean base URL self.base_url = self.base_url.rstrip('/') logger.info("JobAPIService initialized successfully") def call_job_api( self, endpoint: str, method: str = 'GET', data: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, timeout: int = 30 ) -> Dict[str, Any]: """ Make API calls to third-party job service with basic authentication Args: endpoint: API endpoint (without leading slash) method: HTTP method (GET, POST, PUT, PATCH, DELETE) data: Request data/parameters headers: Additional headers timeout: Request timeout in seconds Returns: Dict containing success status, data, and response information """ try: # Clean endpoint endpoint = endpoint.lstrip('/') url = f"{self.base_url}/{endpoint}" # Prepare headers request_headers = headers or {} # Add content type for JSON requests if not specified if method.upper() in ['POST', 'PUT', 'PATCH'] and 'Content-Type' not in request_headers: request_headers['Content-Type'] = 'application/json' # Prepare authentication auth = (self.username, self.password) # Prepare request parameters request_params = { 'auth': auth, 'timeout': timeout, 'headers': request_headers } # Add data based on method if method.upper() == 'GET': request_params['params'] = data or {} else: request_params['json'] = data or {} # Make the API call response = requests.request(method.upper(), url, **request_params) if response.status_code >= 200 and response.status_code < 300: logger.info(f"Job API call successful - {method.upper()} {url} - Status: {response.status_code}") try: response_data = response.json() except ValueError: # Handle non-JSON responses response_data = response.text return { 'success': True, 'data': response_data, 'status': response.status_code, 'headers': dict(response.headers) } else: logger.warning(f"Job API call failed - {method.upper()} {url} - Status: {response.status_code} - Error: {response.text}") return { 'success': False, 'error': f'Job API call failed with status: {response.status_code}', 'status': response.status_code, 'response': response.text } except requests.exceptions.Timeout: error_msg = f"Request timeout after {timeout} seconds" logger.error(f"Job API timeout - {method.upper()} {endpoint} - {error_msg}") return { 'success': False, 'error': error_msg, 'status': 0 } except requests.exceptions.RequestException as e: error_msg = f"Request exception: {str(e)}" logger.error(f"Job API request exception - {method.upper()} {endpoint} - {error_msg}") return { 'success': False, 'error': error_msg, 'status': 0 } except Exception as e: error_msg = f"Unexpected error: {str(e)}" logger.error(f"Job API unexpected error - {method.upper()} {endpoint} - {error_msg}") return { 'success': False, 'error': error_msg, 'status': 0 } def get_employer_job(self, emp_job_id: str) -> Dict[str, Any]: """ Get employer job details by ID Args: emp_job_id: Employer job ID Returns: Dict containing job details or error information """ endpoint = f"api/employer-job/view/{emp_job_id}" return self.call_job_api( endpoint=endpoint, method='GET' ) def get_employer_jobs_list(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Get list of employer jobs with optional parameters Args: params: Optional query parameters for filtering Returns: Dict containing jobs list or error information """ endpoint = "api/employer-job/list" return self.call_job_api( endpoint=endpoint, method='GET', data=params ) def create_employer_job(self, job_data: Dict[str, Any]) -> Dict[str, Any]: """ Create a new employer job Args: job_data: Job creation data Returns: Dict containing creation result or error information """ endpoint = "api/employer-job/create" return self.call_job_api( endpoint=endpoint, method='POST', data=job_data ) def update_employer_job(self, emp_job_id: str, job_data: Dict[str, Any]) -> Dict[str, Any]: """ Update an existing employer job Args: emp_job_id: Employer job ID job_data: Updated job data Returns: Dict containing update result or error information """ endpoint = f"api/employer-job/update/{emp_job_id}" return self.call_job_api( endpoint=endpoint, method='PUT', data=job_data ) def delete_employer_job(self, emp_job_id: str) -> Dict[str, Any]: """ Delete an employer job Args: emp_job_id: Employer job ID Returns: Dict containing deletion result or error information """ endpoint = f"api/employer-job/delete/{emp_job_id}" return self.call_job_api( endpoint=endpoint, method='DELETE' ) def test_connection(self) -> Dict[str, Any]: """ Test the API connection and credentials Returns: Dict containing connection test results """ try: # Try a simple API call to test connection response = self.call_job_api( endpoint="api/health", # Assuming there's a health check endpoint method='GET', timeout=10 ) return { 'connection_test': 'success' if response.get('success') else 'failed', 'base_url': self.base_url, 'username': self.username, 'response': response } except Exception as e: return { 'connection_test': 'failed', 'base_url': self.base_url, 'username': self.username, 'error': str(e) }