140 lines
6.6 KiB
Python
140 lines
6.6 KiB
Python
from requests.exceptions import ConnectTimeout, ReadTimeout
|
||
from tenacity import retry, stop_after_attempt
|
||
from tenacity.wait import wait_exponential
|
||
|
||
from src.constants import FETCH_JP_NAME
|
||
from src.environments import (CRM_AUTH_TIMEOUT,
|
||
CRM_FETCH_RECORD_MAX_RETRY_ATTEMPT,
|
||
CRM_FETCH_RECORD_RETRY_INTERVAL,
|
||
CRM_FETCH_RECORD_RETRY_MAX_INTERVAL,
|
||
CRM_FETCH_RECORD_RETRY_MIN_INTERVAL,
|
||
CRM_FETCH_RECORD_TIMEOUT,
|
||
CRM_GET_RECORD_COUNT_MAX_RETRY_ATTEMPT,
|
||
CRM_GET_RECORD_COUNT_RETRY_INTERVAL,
|
||
CRM_GET_RECORD_COUNT_RETRY_MAX_INTERVAL,
|
||
CRM_GET_RECORD_COUNT_RETRY_MIN_INTERVAL,
|
||
CRM_GET_RECORD_COUNT_TIMEOUT)
|
||
from src.error.exceptions import DataConvertException, SalesforceAPIException
|
||
from src.salesforce.salesforce_api import SalesForceCount, SalesForceData
|
||
from src.util.logger import logger_instance as logger
|
||
|
||
|
||
def fetch_crm_data(target_object, last_fetch_datetime):
|
||
# ① CRMデータ取得処理開始ログを出力する
|
||
logger.info(
|
||
f'I-FETCH-01 [{target_object.object_name}] のCRMからのデータ取得処理を開始します')
|
||
|
||
object_name = target_object.object_name
|
||
columns = ','.join(target_object.columns)
|
||
last_fetch_datetime_from = last_fetch_datetime.last_fetch_datetime_from
|
||
last_fetch_datetime_to = last_fetch_datetime.last_fetch_datetime_to
|
||
|
||
global count_contime_counter, count_readtime_counter, count_counter, data_contime_counter, data_readtime_counter, data_counter
|
||
count_contime_counter = 1
|
||
count_readtime_counter = 1
|
||
count_counter = 1
|
||
data_contime_counter = 1
|
||
data_readtime_counter = 1
|
||
data_counter = 1
|
||
|
||
try:
|
||
# ② 取得対象オブジェクトの取得期間内のレコード件数を取得する
|
||
logger.info(f'I-FETCH-02 [{object_name}] の件数取得を開始します')
|
||
|
||
record_count = fetch_sf_count_retry(
|
||
object_name, last_fetch_datetime_from, last_fetch_datetime_to)
|
||
|
||
logger.info(f'I-FETCH-03 [{object_name}] の件数：[{record_count}]')
|
||
|
||
except Exception as e:
|
||
raise SalesforceAPIException(
|
||
'E-FETCH-01', FETCH_JP_NAME, f'[{object_name}] の件数取得に失敗しました エラー内容：[{e}]')
|
||
|
||
try:
|
||
# ③ 取得対象オブジェクトのレコードを取得する
|
||
logger.info(f'I-FETCH-04 [{object_name}] のレコード取得を開始します')
|
||
|
||
record_generator = fetch_sf_data_retry(
|
||
columns, object_name, last_fetch_datetime_from, last_fetch_datetime_to)
|
||
|
||
except Exception as e:
|
||
raise SalesforceAPIException(
|
||
'E-FETCH-02', FETCH_JP_NAME, f'[{object_name}] のレコード取得に失敗しました エラー内容：[{e}]')
|
||
|
||
try:
|
||
# ④ 取得対象オブジェクトをJSONに変換
|
||
logger.info(f'I-FETCH-05 [{object_name}] のレコードをJSONに変換します')
|
||
|
||
sf_object_jsons = []
|
||
|
||
for record in record_generator:
|
||
sf_object_jsons.append(record)
|
||
|
||
except Exception as e:
|
||
raise DataConvertException(
|
||
'E-FETCH-03', FETCH_JP_NAME, f'[{object_name}] のレコードのJSON変換に失敗しました エラー内容：[{e}]')
|
||
|
||
# ⑤ CRMデータ取得処理終了ログを出力する
|
||
logger.info(f'I-FETCH-06 [{object_name}] のレコード取得が成功しました')
|
||
|
||
# ⑥ 次の処理へ移行する
|
||
return sf_object_jsons
|
||
|
||
|
||
@retry(wait=wait_exponential(multiplier=CRM_GET_RECORD_COUNT_RETRY_INTERVAL, min=CRM_GET_RECORD_COUNT_RETRY_MIN_INTERVAL, max=CRM_GET_RECORD_COUNT_RETRY_MAX_INTERVAL), stop=stop_after_attempt(CRM_GET_RECORD_COUNT_MAX_RETRY_ATTEMPT))
|
||
def fetch_sf_count_retry(object_name, last_update_datetime_from, last_update_datetime_to):
|
||
try:
|
||
global count_contime_counter, count_readtime_counter, count_counter
|
||
|
||
saleforce_count = SalesForceCount()
|
||
return saleforce_count.fetch_sf_count(object_name, last_update_datetime_from, last_update_datetime_to)
|
||
|
||
except ConnectTimeout as e:
|
||
if count_contime_counter < CRM_GET_RECORD_COUNT_MAX_RETRY_ATTEMPT:
|
||
count_contime_counter += 1
|
||
logger.warn(f'W-FETCH-01 CRMの接続処理がタイムアウトしため、リトライします：[{CRM_AUTH_TIMEOUT}] エラー内容：[{e}]')
|
||
raise e
|
||
|
||
except ReadTimeout as e:
|
||
if count_readtime_counter < CRM_GET_RECORD_COUNT_MAX_RETRY_ATTEMPT:
|
||
count_readtime_counter += 1
|
||
logger.warn(
|
||
f'W-FETCH-02 [{object_name}] の件数取得処理がタイムアウトしたため、リトライします：[{CRM_GET_RECORD_COUNT_TIMEOUT}] エラー内容：[{e}]')
|
||
raise e
|
||
|
||
except Exception as e:
|
||
if count_counter < CRM_GET_RECORD_COUNT_MAX_RETRY_ATTEMPT:
|
||
count_counter += 1
|
||
logger.warn(
|
||
f'W-FETCH-03 [{object_name}] の件数取得に失敗したため、リトライします エラー内容：[{e}]')
|
||
raise e
|
||
|
||
|
||
@retry(wait=wait_exponential(multiplier=CRM_FETCH_RECORD_RETRY_INTERVAL, min=CRM_FETCH_RECORD_RETRY_MIN_INTERVAL, max=CRM_FETCH_RECORD_RETRY_MAX_INTERVAL), stop=stop_after_attempt(CRM_FETCH_RECORD_MAX_RETRY_ATTEMPT))
|
||
def fetch_sf_data_retry(columns, object_name, last_update_datetime_from, last_update_datetime_to):
|
||
try:
|
||
global data_contime_counter, data_readtime_counter, data_counter
|
||
|
||
saleforce_data = SalesForceData()
|
||
return saleforce_data.fetch_sf_data(columns, object_name, last_update_datetime_from, last_update_datetime_to)
|
||
|
||
except ConnectTimeout as e:
|
||
if data_contime_counter < CRM_FETCH_RECORD_MAX_RETRY_ATTEMPT:
|
||
data_contime_counter += 1
|
||
logger.warn(f'W-FETCH-04 CRMの接続処理がタイムアウトしため、リトライします：[{CRM_AUTH_TIMEOUT}] エラー内容：[{e}]')
|
||
raise e
|
||
|
||
except ReadTimeout as e:
|
||
if data_readtime_counter < CRM_FETCH_RECORD_MAX_RETRY_ATTEMPT:
|
||
data_readtime_counter += 1
|
||
logger.warn(
|
||
f'W-FETCH-05 [{object_name}] のレコード取得処理がタイムアウトしたため、リトライします：[{CRM_FETCH_RECORD_TIMEOUT}] エラー内容：[{e}]')
|
||
raise e
|
||
|
||
except Exception as e:
|
||
if data_counter < CRM_FETCH_RECORD_MAX_RETRY_ATTEMPT:
|
||
data_counter += 1
|
||
logger.warn(
|
||
f'W-FETCH-06 [{object_name}] のレコード取得に失敗したため、リトライします エラー内容：[{e}]')
|
||
raise e
|