29 lines
1.5 KiB
Python
29 lines
1.5 KiB
Python
from simple_salesforce import Salesforce
|
|
from src.system_var.environments import (CRM_AUTH_DOMAIN, CRM_AUTH_TIMEOUT,
|
|
CRM_FETCH_RECORD_TIMEOUT,
|
|
CRM_GET_RECORD_COUNT_TIMEOUT,
|
|
CRM_USER_NAME, CRM_USER_PASSWORD,
|
|
CRM_USER_SECURITY_TOKEN)
|
|
|
|
|
|
class SalesforceApiClient():
|
|
def __init__(self) -> None:
|
|
self.__sf = Salesforce(username=CRM_USER_NAME, password=CRM_USER_PASSWORD,
|
|
security_token=CRM_USER_SECURITY_TOKEN,
|
|
domain=CRM_AUTH_DOMAIN
|
|
)
|
|
|
|
def __query(self, soql, include_deleted=True, conn_timeout=100, read_timeout=300):
|
|
return self.__sf.query(soql, include_deleted, timeout=(float(conn_timeout), float(read_timeout)))
|
|
|
|
def __query_all(self, soql, include_deleted=True, conn_timeout=100, read_timeout=300):
|
|
return self.__sf.query_all(soql, include_deleted, timeout=(float(conn_timeout), float(read_timeout)))
|
|
|
|
def fetch_sf_count(self, soql: str):
|
|
count_res = self.__query(soql, conn_timeout=CRM_AUTH_TIMEOUT, read_timeout=CRM_GET_RECORD_COUNT_TIMEOUT)
|
|
return count_res.get('records')[0].get('expr0')
|
|
|
|
def fetch_sf_data(self, soql: str):
|
|
data_res = self.__query_all(soql, conn_timeout=CRM_AUTH_TIMEOUT, read_timeout=CRM_FETCH_RECORD_TIMEOUT)
|
|
return data_res.get('records')
|