forked from DigiFi/digifi-BankToProductCore
VivianDee
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
import httpx
|
|
import json
|
|
from app.utils.logger import logger
|
|
from app.config import settings
|
|
import logging
|
|
|
|
|
|
class SimbrellaIntegration:
|
|
BASE_URL = settings.SIMBRELLA_BASE_URL
|
|
ENDPOINT_RAC_CHECKS = settings.SIMBRELLA_ENDPOINT_RAC_CHECKS
|
|
HEALTH_ENDPOINT = settings.SIMBRELLA_HEALTH
|
|
AUTH_ENDPOINT = settings.BANK_CALL_AUTH_ENDPOINT
|
|
|
|
_access_token = None # cache token in memory
|
|
|
|
@staticmethod
|
|
def generate_token():
|
|
"""
|
|
Generate a new access token using the username and password from settings.
|
|
"""
|
|
url = f"{SimbrellaIntegration.BASE_URL}{SimbrellaIntegration.AUTH_ENDPOINT}"
|
|
|
|
payload = {
|
|
"username": settings.BANK_CALL_USERNAME,
|
|
"password": settings.BANK_CALL_PASSWORD,
|
|
"grant_type": "password"
|
|
}
|
|
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
try:
|
|
logger.info(f"Requesting token from {url}")
|
|
|
|
response = httpx.post(url, json=payload, headers=headers, timeout=10.0)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
SimbrellaIntegration._access_token = data.get("access_token")
|
|
|
|
if not SimbrellaIntegration._access_token:
|
|
raise Exception("Access token not found in response")
|
|
|
|
logger.info("Successfully retrieved access token")
|
|
return SimbrellaIntegration._access_token
|
|
|
|
except Exception as e:
|
|
logger.error(f"Token generation failed: {str(e)}", exc_info=True)
|
|
raise Exception(f"Token generation failed: {str(e)}")
|
|
|
|
|
|
@staticmethod
|
|
def rac_check(customer_id, account_id, transaction_id):
|
|
"""
|
|
Calls the RACCheck endpoit
|
|
"""
|
|
url = f"{SimbrellaIntegration.BASE_URL}/{SimbrellaIntegration.ENDPOINT_RAC_CHECKS}"
|
|
logger.info(f"Contacting Rack Checks EndPoint: {str(url)}", exc_info=True)
|
|
|
|
payload = {
|
|
"customerId": customer_id,
|
|
"accountId": account_id,
|
|
"transactionId": str(transaction_id),
|
|
"fbnTransactionId": str(transaction_id),
|
|
"countryCode": "NG",
|
|
"channel": "USSD"
|
|
}
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {SimbrellaIntegration._access_token}"
|
|
}
|
|
|
|
try:
|
|
response = httpx.post(url, json=payload, headers=headers, timeout=10.0)
|
|
|
|
logger.info(f"This is Response: {str(response)}", exc_info=True)
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
logger.error(f"RACCheck API call failed: {str(e)}", exc_info=True)
|
|
raise Exception(f"RACCheck API call failed: {str(e)}")
|
|
|
|
@staticmethod
|
|
def health_check():
|
|
"""
|
|
Health check for Simbrella Service
|
|
"""
|
|
|
|
url = f"{SimbrellaIntegration.BASE_URL}/{SimbrellaIntegration.HEALTH_ENDPOINT}"
|
|
logger.info(f"Simbrella Health Check URL: {url}")
|
|
|
|
try:
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {SimbrellaIntegration._access_token}"
|
|
}
|
|
|
|
response = httpx.get(url, headers=headers, timeout=10.0)
|
|
logger.info(f"Simbrella Health Check Response: {response.text}")
|
|
return response
|
|
except Exception as e:
|
|
logger.error(f"Simbrella Health Check API call failed: {str(e)}", exc_info=True)
|
|
raise Exception(f"Simbrella Health Check API call failed: {str(e)}")
|
|
|