148 lines
6.6 KiB
Python
148 lines
6.6 KiB
Python
from src.system_var.constants import (COLUMNS_KEY, COLUMNS_TYPE,
|
|
DATE_PATTERN_EXPECTED_YYYYMMDDTHHMMSSTZ,
|
|
DATE_PATTERN_YYYYMMDDTHHMMSSTZ,
|
|
DATETIME_COLUMN_DEFAULT_VALUE,
|
|
DATETIME_COLUMN_KEY,
|
|
DATETIME_COLUMN_TYPE, IS_SKIP_KEY,
|
|
IS_SKIP_TYPE,
|
|
IS_UPDATE_LAST_FETCH_DATETIME_KEY,
|
|
IS_UPDATE_LAST_FETCH_DATETIME_TYPE,
|
|
LAST_FETCH_DATETIME_FILE_NAME_KEY,
|
|
LAST_FETCH_DATETIME_FILE_NAME_TYPE,
|
|
LAST_FETCH_DATETIME_FROM_KEY,
|
|
LAST_FETCH_DATETIME_FROM_TYPE,
|
|
LAST_FETCH_DATETIME_TO_KEY,
|
|
LAST_FETCH_DATETIME_TO_TYPE,
|
|
OBJECT_NAME_KEY, OBJECT_NAME_TYPE,
|
|
OBJECTS_KEY, OBJECTS_TYPE,
|
|
UPLOAD_FILE_NAME_KEY,
|
|
UPLOAD_FILE_NAME_TYPE)
|
|
from src.util.dict_checker import DictChecker
|
|
from src.util.execute_datetime import ExecuteDateTime
|
|
|
|
|
|
class FetchTargetObjects():
|
|
def __init__(self, object_info_file_dict) -> None:
|
|
self.__objects = object_info_file_dict
|
|
self.__dict_checker = DictChecker(self.__objects)
|
|
self.__validate()
|
|
self.__i = 0
|
|
|
|
def __iter__(self):
|
|
return self
|
|
|
|
def __next__(self):
|
|
if self.__i == len(self.__objects[OBJECTS_KEY]):
|
|
raise StopIteration()
|
|
value = self.__objects[OBJECTS_KEY][self.__i]
|
|
self.__i += 1
|
|
return value
|
|
|
|
def __validate(self) -> None:
|
|
self.__dict_checker.assert_key_exist(OBJECTS_KEY)
|
|
self.__dict_checker.assert_data_type(OBJECTS_KEY, OBJECTS_TYPE)
|
|
|
|
|
|
class TargetObject():
|
|
def __init__(self, object_info, execute_datetime: ExecuteDateTime) -> None:
|
|
self.__dict_checker = DictChecker(object_info)
|
|
self.__object_info = object_info
|
|
self.__execute_datetime = execute_datetime
|
|
self.__validate()
|
|
|
|
def __validate(self) -> None:
|
|
self.__validate_required_properties()
|
|
self.__validate_optional_properties()
|
|
|
|
return
|
|
|
|
def __validate_required_properties(self) -> None:
|
|
self.__dict_checker.assert_key_exist(OBJECT_NAME_KEY)
|
|
self.__dict_checker.assert_data_type(OBJECT_NAME_KEY, OBJECT_NAME_TYPE)
|
|
self.__dict_checker.assert_key_exist(COLUMNS_KEY)
|
|
self.__dict_checker.assert_data_type(COLUMNS_KEY, COLUMNS_TYPE)
|
|
self.__dict_checker.assert_list_empty(COLUMNS_KEY)
|
|
|
|
return
|
|
|
|
def __validate_optional_properties(self) -> None:
|
|
if self.__dict_checker.check_key_exist(IS_SKIP_KEY):
|
|
self.__dict_checker.assert_data_type(IS_SKIP_KEY, IS_SKIP_TYPE)
|
|
|
|
if self.__dict_checker.check_key_exist(IS_UPDATE_LAST_FETCH_DATETIME_KEY):
|
|
self.__dict_checker.assert_data_type(IS_UPDATE_LAST_FETCH_DATETIME_KEY, IS_UPDATE_LAST_FETCH_DATETIME_TYPE)
|
|
|
|
if self.__dict_checker.check_key_exist(LAST_FETCH_DATETIME_FILE_NAME_KEY):
|
|
self.__dict_checker.assert_data_type(LAST_FETCH_DATETIME_FILE_NAME_KEY, LAST_FETCH_DATETIME_FILE_NAME_TYPE)
|
|
|
|
if self.__dict_checker.check_key_exist(UPLOAD_FILE_NAME_KEY):
|
|
self.__dict_checker.assert_data_type(UPLOAD_FILE_NAME_KEY, UPLOAD_FILE_NAME_TYPE)
|
|
|
|
if self.__dict_checker.check_key_exist(DATETIME_COLUMN_KEY):
|
|
self.__dict_checker.assert_data_type(DATETIME_COLUMN_KEY, DATETIME_COLUMN_TYPE)
|
|
|
|
return
|
|
|
|
@property
|
|
def object_name(self) -> str:
|
|
return self.__object_info[OBJECT_NAME_KEY]
|
|
|
|
@property
|
|
def columns(self) -> list:
|
|
return self.__object_info[COLUMNS_KEY]
|
|
|
|
@property
|
|
def is_skip(self) -> bool:
|
|
return self.__object_info[IS_SKIP_KEY] if self.__dict_checker.check_key_exist(IS_SKIP_KEY) else False
|
|
|
|
@property
|
|
def is_update_last_fetch_datetime(self) -> bool:
|
|
if self.__dict_checker.check_key_exist(IS_UPDATE_LAST_FETCH_DATETIME_KEY):
|
|
return self.__object_info[IS_UPDATE_LAST_FETCH_DATETIME_KEY]
|
|
return True
|
|
|
|
@property
|
|
def last_fetch_datetime_file_name(self) -> str:
|
|
if self.__dict_checker.check_key_exist(LAST_FETCH_DATETIME_FILE_NAME_KEY):
|
|
return self.__object_info[LAST_FETCH_DATETIME_FILE_NAME_KEY]
|
|
return f'{self.__object_info[OBJECT_NAME_KEY]}.json'
|
|
|
|
@property
|
|
def upload_file_name(self) -> str:
|
|
if self.__dict_checker.check_key_exist(UPLOAD_FILE_NAME_KEY):
|
|
return self.__object_info[UPLOAD_FILE_NAME_KEY].format(execute_datetime=self.__execute_datetime.format_date())
|
|
return f'CRM_{self.__object_info[OBJECT_NAME_KEY]}_{self.__execute_datetime.format_date()}'
|
|
|
|
@property
|
|
def datetime_column(self) -> str:
|
|
return self.__object_info[DATETIME_COLUMN_KEY] if self.__dict_checker.check_key_exist(DATETIME_COLUMN_KEY) else DATETIME_COLUMN_DEFAULT_VALUE
|
|
|
|
|
|
class LastFetchDatetime():
|
|
def __init__(self, last_fetch_datetime_file_dict, execute_datetime) -> None:
|
|
self.__dict_checker = DictChecker(last_fetch_datetime_file_dict)
|
|
self.__execute_datetime = execute_datetime
|
|
self.__last_fetch_datetime_file_dict = last_fetch_datetime_file_dict
|
|
self.__validate()
|
|
|
|
def __validate(self) -> None:
|
|
self.__dict_checker.assert_key_exist(LAST_FETCH_DATETIME_FROM_KEY)
|
|
self.__dict_checker.assert_data_type(LAST_FETCH_DATETIME_FROM_KEY, LAST_FETCH_DATETIME_FROM_TYPE)
|
|
self.__dict_checker.assert_match_pattern(LAST_FETCH_DATETIME_FROM_KEY, DATE_PATTERN_YYYYMMDDTHHMMSSTZ,
|
|
DATE_PATTERN_EXPECTED_YYYYMMDDTHHMMSSTZ)
|
|
if self.__dict_checker.check_key_exist(LAST_FETCH_DATETIME_TO_KEY):
|
|
self.__dict_checker.assert_data_type(LAST_FETCH_DATETIME_TO_KEY, LAST_FETCH_DATETIME_TO_TYPE)
|
|
self.__dict_checker.assert_match_pattern(LAST_FETCH_DATETIME_TO_KEY, DATE_PATTERN_YYYYMMDDTHHMMSSTZ,
|
|
DATE_PATTERN_EXPECTED_YYYYMMDDTHHMMSSTZ)
|
|
return
|
|
|
|
@property
|
|
def last_fetch_datetime_from(self) -> str:
|
|
return self.__last_fetch_datetime_file_dict[LAST_FETCH_DATETIME_FROM_KEY]
|
|
|
|
@property
|
|
def last_fetch_datetime_to(self) -> str:
|
|
if self.__dict_checker.check_key_exist(LAST_FETCH_DATETIME_TO_KEY):
|
|
return self.__last_fetch_datetime_file_dict[LAST_FETCH_DATETIME_TO_KEY]
|
|
return str(self.__execute_datetime)
|