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

コメント整理

Browse Source
This commit is contained in:
x.azuma.m@nds-tyo.co.jp 2023-07-14 12:54:59 +09:00
parent ecbff882bf
commit 241838ef71
4 changed files with 89 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
ecs/jskult-batch-daily/src/batch/vjsk/vjsk_data_load_manager.py
View File

@ -11,10 +11,21 @@ mapper = VjskReceiveFileMapper()
class VjskDataLoadManager:
"""
Ｖ実消化データ取込機能クラス
"""
def __init__(self):
pass
def _import_to_db(src_file_name: str, condkey: str):
"""
概要
指定されたtsvファイル src_file_name を策席スキーマに登録する
引数
src_file_name: ローカルストレージにある取込対象tsvファイルパス
condkey: 受領データの種類を一意に示す値(VjskReceiveFileMapperクラスで管理されているCONDKEY値)
"""
logger.debug(f"_import_to_db start (src_file_name : {src_file_name}, condkey : {condkey})")
db = Database.get_instance()
@ -24,6 +35,7 @@ class VjskDataLoadManager:
upsert_sql = mapper.get_upsert_sql(condkey)
try:
# データベース接続
db.connect()
db.execute("SET SESSION sql_mode = 'TRADITIONAL';")
@ -52,17 +64,27 @@ class VjskDataLoadManager:
# MEMO: https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.BaseCursorResult.rowcount
logger.info(f'{table_name_org}を{table_name_src}にUPSERT')
# データベースコミット
db.commit()
except Exception as e:
db.rollback()
raise BatchOperationException(e)
finally:
# データベース切断
db.disconnect()
logger.debug("_import_to_db done")
return
def _get_tsv_last_row_tab_count(src_file_name: str) -> int:
"""
概要
指定されたtsvファイル src_file_name の末尾行に含まれるタブ文字数を取得する
引数
src_file_name: ローカルストレージにある取込対象tsvファイルパス
"""
# memo: tsvファイルが数百MBに及ぶことを想定して、末尾から1行分を参照する
# memo: 前提1 行区切りは LF('\n')
# memo: 前提2 正常時のファイル終端にある文字は、末尾行の LF('\n')
@ -103,12 +125,25 @@ class VjskDataLoadManager:
@classmethod
def load(self, target: dict):
"""
概要
取込対象受領ファイル target をデータベースに登録する
引数
target: {
condkey: 受領データの種類を一意に示す値(VjskReceiveFileMapperクラスで管理されているCONDKEY値)
src_file_path: ローカルストレージにある取込対象tsvファイルパス
}
"""
logger.debug(f'load start target:{target}')
# S3からローカルストレージにdownloadした登録対象のtsvファイルパスを取得
local_file_name = target["src_file_path"]
# tsvファイル末尾行のTABの数が総定数と一致しない場合は例外をスロー
# memo:
# 対向元システムで生成されるファイルは稀に途中欠落が発生することがある。
# これを、ファイルMySQL8.0のLOADステートメントで発生するWARNING/ERRORでは検知不可能なので、
# LOADステートメント実行前に、物理的に途中欠落があるかを検知してエラーとすることが目的。
tsv_tabs = self._get_tsv_last_row_tab_count(local_file_name)
expect_tabs = mapper.get_file_column_separators(target["condkey"])
if tsv_tabs != expect_tabs:

51
ecs/jskult-batch-daily/src/batch/vjsk/vjsk_recv_file_mapper.py
View File

@ -2,6 +2,9 @@ import textwrap
class VjskReceiveFileMapper:
"""
Ｖ実消化データファイルI/Fマッピング定義クラス
"""
CONDKEY_SLIP_DATA = "SLIP_DATA" # 販売実績データ
CONDKEY_HLD_MST = "HLD_MST" # Ｖ卸ホールディングスマスタ
CONDKEY_WHS_MST = "WHS_MST" # Ｖ卸マスタ
@ -1489,27 +1492,75 @@ class VjskReceiveFileMapper:
return ret
def get_data_name(self, condkey: str) -> str:
"""
概要
受領ファイルI/Fが想定する、ファイル論理名を取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return self._get_interface_property(condkey, self._KEY_DATA_NAME)
def get_file_prefix(self, condkey: str) -> str:
"""
概要
受領ファイルI/Fが想定する、ファイル名接頭辞を取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return self._get_interface_property(condkey, self._KEY_FILE_PREFIX)
def get_file_suffix(self, condkey: str) -> str:
"""
概要
受領ファイルI/Fが想定する、ファイル拡張子を取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return self._get_interface_property(condkey, self._KEY_FILE_SUFFIX)
def get_file_column_separators(self, condkey: str) -> int:
"""
概要
受領ファイルI/Fが想定する、1行あたりのタブ文字数を取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return int(self._get_interface_property(condkey, self._KEY_FILE_COLUMN_SEPARATORS))
def get_org_table(self, condkey: str) -> str:
"""
概要
受領ファイルI/Fが想定する、LOAD先ロードスキーマテーブル名を取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return self._get_interface_property(condkey, self._KEY_ORG_TABLE)
def get_src_table(self, condkey: str) -> str:
"""
概要
受領ファイルI/Fが想定する、登録先蓄積スキーマテーブル名を取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return self._get_interface_property(condkey, self._KEY_SRC_TABLE)
def get_upsert_sql(self, condkey: str) -> str:
"""
概要
受領ファイルI/Fが想定する、upsert (ロードスキーマ→蓄積スキーマ) CMLステートメントを取得する
引数
condkey: 受領データの種類を一意に示す値(このクラスのメンバ CONDKEY_* の値)
"""
return self._get_interface_property(condkey, self._KEY_UPSERT_SQL)
def get_condkey_by_s3_file_path(self, s3_file_path: str) -> str:
"""
概要
S3受領バケットに受領したファイル名から、I/F想定に該当する condkey 値を取得する
引数
s3_file_path: S3受領バケットにある受領したファイルパス
"""
ret = None
filename = s3_file_path[s3_file_path.rfind("/") + 1:]
for condkey in self._VJSK_INTERFACE_MAPPING:

1
ecs/jskult-batch-daily/tests/batch/vjsk/vjsk_file_check/conftest.py
View File

@ -23,7 +23,6 @@ def receive_folder():
return os.environ["VJSK_DATA_RECEIVE_FOLDER"]
# TODO 共通fixtureにして15個固定でput/delete、各個別fixtureで15個から引き算でdeleteする
@pytest.fixture
def init_check_received_files_ok1(s3_client, bucket_name, receive_folder):
# setup

3
ecs/jskult-batch-daily/tests/batch/vjsk/vjsk_load/test_vjsk_load.py
View File

@ -17,6 +17,9 @@ from tests.testing_vjsk_utility import (assert_table_results,
class TestImportFileToDb:
"""
Ｖ実消化データ取込－データベース登録処理 テストクラス
"""
db: Database
batch_context: BatchContext
test_file_path_import_all: str