nik/newdwh2021
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: アルトマーク取込でバッチステータス管理テーブルへの書き込み処理を追加

Browse Source
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2025-05-28 11:34:20 +09:00
parent c475ad9008
commit 6d699c8cc0
4 changed files with 495 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ecs/jskult-batch-ultmarc-io/src/error/exceptions.py
View File

@ -8,3 +8,7 @@ class DBException(MeDaCaException):
class BatchOperationException(MeDaCaException):
pass
class MaxRunCountReachedException(MeDaCaException):
pass

30
ecs/jskult-batch-ultmarc-io/src/main.py
View File

@ -4,6 +4,7 @@ from src.batch.common.batch_context import BatchContext
from src.batch.ultmarc import import_ultmarc_process, output_dcf_dsf_data
from src.error.exceptions import BatchOperationException
from src.logging.get_logger import get_logger
from src.manager.jskult_batch_status_manager import JskultBatchStatusManager
from src.manager.jskult_hdke_tbl_manager import JskultHdkeTblManager
from src.system_var import constants
@ -18,13 +19,22 @@ def exec():
logger.info('アルトマーク取込/データ出力：開始')
hdke_tbl_manager = JskultHdkeTblManager()
batch_status_manager = JskultBatchStatusManager(
constants.PROCESS_NAME_ULTMARC_IO,
constants.PROCESS_TYPE_DATA_IMPORT,
0, # 最大起動回数は使用しないため、0をセット
0, # 受信ファイル数は使用しないため、0をセット
)
# バッチステータスを処理開始に変更
batch_status_manager.set_process_status(constants.PROCESS_STATUS_START)
try:
if not hdke_tbl_manager.can_run_process():
logger.error('日次バッチ処理中またはdump取得が正常終了していないため、日次バッチ処理を終了します。')
return constants.BATCH_EXIT_CODE_SUCCESS
except BatchOperationException as e:
logger.exception(f'日付テーブルチェック処理エラー（異常終了）{e}')
# TODO: バッチステータス管理テーブルにエラーを登録
# バッチステータスをエラーに変更
batch_status_manager.set_process_status(constants.PROCESS_STATUS_ERROR)
return constants.BATCH_EXIT_CODE_SUCCESS
_, _, syor_date = hdke_tbl_manager.get_batch_statuses()
@ -32,7 +42,8 @@ def exec():
# バッチ共通設定に処理日を追加
batch_context.syor_date = syor_date
# TODO: バッチステータス管理テーブルに処理中を登録
# バッチステータスを処理中を変更
batch_status_manager.set_process_status(constants.PROCESS_STATUS_DOING)
try:
logger.info('アルトマーク取込：起動')
@ -40,19 +51,22 @@ def exec():
logger.info('アルトマーク取込：終了')
except BatchOperationException as e:
logger.exception(f'アルトマーク取込処理エラー（異常終了）{e}')
# TODO: バッチステータス管理テーブルにエラーを登録
# バッチステータスをエラーに変更
batch_status_manager.set_process_status(constants.PROCESS_STATUS_ERROR)
return constants.BATCH_EXIT_CODE_SUCCESS
try:
logger.info('V実消化用施設データ作成処理：起動')
logger.info('実消化用DCF/DSFデータ作成処理：起動')
output_dcf_dsf_data.exec()
logger.info('V実消化用施設データ作成処理：終了')
logger.info('実消化用DCF/DSFデータ作成処理：終了')
except BatchOperationException as e:
logger.exception(f'V実消化用施設データ作成処理エラー（異常終了）{e}')
# TODO: バッチステータス管理テーブルにエラーを登録
logger.exception(f'実消化用施設DCF/DSF作成処理エラー（異常終了）{e}')
# バッチステータスをエラーに変更
batch_status_manager.set_process_status(constants.PROCESS_STATUS_ERROR)
return constants.BATCH_EXIT_CODE_SUCCESS
# TODO: バッチステータス管理テーブルに処理済を登録
# バッチステータスを処理済に変更
batch_status_manager.set_process_status(constants.PROCESS_STATUS_DONE)
logger.info('アルトマーク取込/データ出力：終了')
return constants.BATCH_EXIT_CODE_SUCCESS

450
ecs/jskult-batch-ultmarc-io/src/manager/jskult_batch_status_manager.py Normal file
View File

@ -0,0 +1,450 @@
from src.db.database import Database
from src.error.exceptions import (BatchOperationException,
MaxRunCountReachedException)
from src.system_var import constants
class JskultBatchStatusManager:
"""実消化&アルトマーク_バッチステータス管理テーブルを管理するクラス"""
def __init__(self, process_name: str, process_type: str, max_run_count: int, receive_file_count: int):
"""コンストラクタ
Args:
process_name (str): 処理名
process_type (str): 管理区分
max_run_count (int): 最大起動回数
receive_file_count (int): 受信ファイル数
"""
self._process_name: str = process_name
self._process_type: str = process_type
self._max_run_count: str = max_run_count
self._receive_file_count: str = receive_file_count
# DB接続モジュールを初期化
self._db = Database.get_instance()
# 処理ステータスの登録および更新を行う
def set_process_status(self, process_status: str):
"""
処理ステータスの登録および更新を行う
Args:
process_status (str): 処理ステータス
Raises:
BatchOperationException: DB操作の何らかのエラー
"""
try:
self._db.connect()
self._db.begin()
self._db.to_jst()
self._db.execute(
"""
CALL
internal07.upsert_jskult_batch_status_manage(
:process_name,
:process_type,
:process_status,
NULL,
NULL
);""",
{
'process_name': self._process_name,
'process_type': self._process_type,
'process_status': process_status,
}
)
self._db.commit()
except Exception as e:
# 例外発生時はrollback
self._db.rollback()
raise BatchOperationException(e)
finally:
self._db.disconnect()
def can_run_post_process(self) -> bool:
"""
後続処理を実行してよいか判定する
Raises:
BatchOperationException: DB操作の何らかのエラー
MaxRunCountReachedException: 最大起動回数到達時の例外
Returns:
bool 後続処理を実行してよい場合はTrue
"""
try:
self._db.connect()
# 自身（後続処理 or 日付テーブル更新）のレコードを取得する
record = self._db.execute_select(
"""
SELECT
process_name,
process_date
FROM
internal07.jskult_batch_status_manage
WHERE
process_name = :process_name
AND process_date = src07.get_syor_date();
""",
{'process_name': self._process_name}
)
record_count = len(record)
if record_count == 0:
raise BatchOperationException("レコードの取得が出来ませんでした。")
# 起動回数のインクリメント
self._increment_run_count()
# データ取込が完了していた場合
if self._is_done_data_import():
return True
# 最大起動回数に到達していない場合
if not self._is_max_run_count_reached():
return False
# 最大起動回数に到達していた場合
# 最大起動回数フラグを立てて例外を送出する
self._activate_max_run_count_flg()
raise MaxRunCountReachedException("最大起動回数に到達しました")
except MaxRunCountReachedException as e:
raise e
except Exception as e:
raise BatchOperationException(e)
finally:
self._db.disconnect()
def can_run_business_day_update(self) -> bool:
"""
日付テーブル更新を実行してよいか判定する
Raises:
BatchOperationException: DB操作の何らかのエラー
MaxRunCountReachedException: 最大起動回数到達時の例外
Returns:
bool: 日付テーブル更新を実行してよい場合はTrue
"""
try:
# 自身（後続処理 or 日付テーブル更新）のレコードを取得する
self._db.connect()
record = self._db.execute_select(
"""
SELECT
process_name,
process_date
FROM
internal07.jskult_batch_status_manage
WHERE
process_name = :process_name
AND process_date = src07.get_syor_date();
""",
{'process_name': self._process_name}
)
record_count = len(record)
if record_count == 0:
raise BatchOperationException("レコードの取得が出来ませんでした。")
# 起動回数のインクリメント
self._increment_run_count()
# 後続処理が完了していた場合
if self._is_done_post_process():
return True
# 最大起動回数に到達していない場合
if not self._is_max_run_count_reached():
return False
# 最大起動回数フラグを立てる
self._activate_max_run_count_flg()
# 最大起動回数に到達した場合にメッセージをスロー
raise MaxRunCountReachedException("最大起動回数に到達しました")
except MaxRunCountReachedException as e:
raise e
except Exception as e:
raise BatchOperationException(e)
finally:
self._db.disconnect()
def is_done_ultmarc_import(self) -> bool:
"""アルトマークデータ連携があったかを確認する
Raises:
BatchOperationException: DB操作の何らかのエラー
Returns:
bool アルトマークデータ連携があった場合はTrue
"""
try:
self._db.connect()
record = self._db.execute_select(
"""
SELECT
process_name,
process_date
FROM
internal07.jskult_batch_status_manage
WHERE
process_name = :process_name
AND process_date = src07.get_syor_date()
AND process_status = :process_status;
""",
{
'process_name': constants.PROCESS_NAME_ULTMARC_IO,
'process_status': constants.PROCESS_STATUS_DONE
}
)
record_count = len(record)
return record_count != 0
except Exception as e:
raise BatchOperationException(e)
finally:
self._db.disconnect()
def _increment_run_count(self):
"""起動回数をインクリメントする
Raises:
BatchOperationException: DB操作の何らかのエラー
"""
try:
self._db.connect()
self._db.begin()
self._db.to_jst()
# 自身（後続処理 or 日付テーブル更新）のレコードを取得する
record = self._db.execute_select(
"""
SELECT
total_run_count
FROM
internal07.jskult_batch_status_manage
WHERE
process_name = :process_name
AND process_date = src07.get_syor_date();
""",
{'process_name': self._process_name}
)
if len(record) == 0:
raise BatchOperationException("レコードの取得が出来ませんでした。")
total_run_count = record[0]['total_run_count']
self._db.execute(
"""
CALL
internal07.upsert_jskult_batch_status_manage(
:process_name,
:process_type,
NULL,
:total_run_count,
NULL
)
""",
{
'process_name': self._process_name,
'process_type': self._process_type,
'total_run_count': total_run_count + 1,
}
)
self._db.commit()
except Exception as e:
# 例外発生時はrollback
self._db.rollback()
raise BatchOperationException(e)
finally:
self._db.disconnect()
def _is_done_data_import(self) -> bool:
"""データ取込処理が完了しているかを判定する
Raises:
BatchOperationException: DB操作の何らかのエラー
MaxRunCountReachedException: 最大起動回数到達時の例外
Returns:
bool 後続処理を実行してよい場合はTrue
"""
try:
self._db.connect()
# 自身（後続処理 or 日付テーブル更新）のレコードを取得する
record = self._db.execute_select(
"""
SELECT
process_name,
process_date
FROM
internal07.jskult_batch_status_manage
WHERE
process_type = :process_type
AND process_status = :process_status
AND process_date = src07.get_syor_date();
""",
{
'process_type': constants.PROCESS_TYPE_DATA_IMPORT,
'process_status': constants.PROCESS_STATUS_DONE
}
)
record_count = len(record)
# データ取込数とデータ登録の件数が一致しているか確認
return self._receive_file_count == record_count
except Exception as e:
raise BatchOperationException(e)
finally:
self._db.disconnect()
def _is_done_post_process(self) -> bool:
"""後続処理のすべての処理が完了しているかを判定する
Returns:
bool: 後続処理がすべて完了していたらTrue
Raises:
BatchOperationException: DB操作の何らかのエラー
"""
# 生物由来データロット分解のチェック
if not self._is_done_process(constants.PROCESS_NAME_TRN_RESULT_DATA_BIO_LOT):
return False
# メルク施設マスタ作成のチェック
if not self._is_done_process(constants.PROCESS_NAME_MST_INST_ALL):
return False
# DCF削除新規マスタ作成のチェック
if not self._is_done_process(constants.PROCESS_NAME_DCF_INST_MERGE_IO):
return False
# 全ての後続処理が完了している場合Trueを返す
return True
# データ取込処理が完了しているかを判定する
def _is_done_process(self, process_name: str) -> bool:
"""指定された処理名の処理が完了しているかを判定する
Args:
process_name (str): 処理名
Returns:
bool: 処理名に一致する処理が完了していたらTrue
Raises:
BatchOperationException: DB操作の何らかのエラー
"""
try:
self._db.connect()
record = self._db.execute_select(
"""
SELECT
process_name,
process_date
FROM
internal07.jskult_batch_status_manage
WHERE
process_name = :process_name
AND process_status = :process_status
AND process_date = src07.get_syor_date();
""",
{
'process_name': process_name,
'process_status': constants.PROCESS_STATUS_DONE
}
)
record_count = len(record)
return record_count != 0
except Exception as e:
raise BatchOperationException(e)
finally:
self._db.disconnect()
def _is_max_run_count_reached(self) -> bool:
"""起動回数が最大回数に到達しているか判定する
Returns:
bool: 最大起動回数に到達していたらTrue
Raises:
BatchOperationException: DB操作の何らかのエラー
"""
try:
self._db.connect()
record = self._db.execute_select(
"""
SELECT
total_run_count
FROM
internal07.jskult_batch_status_manage
WHERE
process_name = :process_name
AND
process_date = src07.get_syor_date();
""",
{'process_name': self._process_name}
)
total_run_count = record[0]['total_run_count']
# 取得した起動回数とフィールド変数の最大起動回数が一致を確認
return total_run_count == self._max_run_count
except Exception as e:
raise BatchOperationException(e)
finally:
self._db.disconnect()
def _activate_max_run_count_flg(self):
"""最大起動回数フラグにフラグを立てる
Raises:
BatchOperationException: DB操作の何らかのエラー
"""
try:
self._db.connect()
self._db.begin()
self._db.to_jst()
self._db.execute(
"""
CALL
internal07.upsert_jskult_batch_status_manage(
:process_name,
:process_type,
NULL,
NULL,
:max_run_count_flag);
""",
{
'process_name': self._process_name,
'process_type': self._process_type,
'max_run_count_flag': constants.MAX_RUN_COUNT_FLAG_ON
}
)
self._db.commit()
except Exception as e:
self._db.rollback()
raise BatchOperationException(e)
finally:
self._db.disconnect()

19
ecs/jskult-batch-ultmarc-io/src/system_var/constants.py
View File

@ -9,3 +9,22 @@ BATCH_ACTF_BATCH_START = '1'
DUMP_STATUS_KBN_UNPROCESSED = '0'
# dump取得状態区分：dump取得正常終了
DUMP_STATUS_KBN_COMPLETE = '2'
# バッチステータス管理:
# 処理名:
PROCESS_NAME_ULTMARC_IO = 'jskult-batch-ultmarc-io'
# 管理区分:
# データ取込
PROCESS_TYPE_DATA_IMPORT = 'data_import'
# 処理ステータス:
# 処理開始
PROCESS_STATUS_START = 'start'
# 処理待
PROCESS_STATUS_WAITING = 'waiting'
# 処理中
PROCESS_STATUS_DOING = 'doing'
# 処理済
PROCESS_STATUS_DONE = 'done'
# エラー
PROCESS_STATUS_ERROR = 'error'