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

Revert "feat: async-awaitで非同期処理を制御できるようにした"

Browse Source 
This reverts commit d8b6dd9978f72eb227e4b4e2fa15ae871691bd71.
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2023-04-19 13:39:33 +09:00
parent d8b6dd9978
commit ac374afb85
3 changed files with 21 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
ecs/jskult-batch-daily/src/batch/bio_sales/create_bio_sales_lot.py
View File

@ -13,11 +13,6 @@ async def exec():
logger.info('営業日ではないため、生物由来卸販売ロット分解処理をスキップします。')
return
# 並列処理の確認用コード
# import asyncio
# for i in range(50):
# logger.debug('生物由来卸販売ロット分解：処理中')
# await asyncio.sleep(0.5)
# TODO: ここに処理を追記していく
logger.debug('生物由来卸販売ロット分解：終了')

43
ecs/jskult-batch-daily/src/batch/laundering/emp_chg_inst_laundering.py
View File

@ -1,52 +1,51 @@
from src.batch.batch_functions import logging_sql
from src.batch.common.batch_context import BatchContext
from src.db.async_database import AsyncDatabase
from src.db.database import Database
from src.error.exceptions import BatchOperationException
from src.logging.get_logger import get_logger
from src.time.elapsed_time import ElapsedTime
logger = get_logger('施設担当者マスタ洗替')
logger = get_logger('48-施設担当者マスタ洗替')
batch_context = BatchContext.get_instance()
async def exec():
db = AsyncDatabase.get_instance()
db = Database.get_instance()
try:
await db.connect()
db.connect()
logger.debug('##########################')
logger.debug('START Changing Employee in charge of institution PGM.')
# 業務日付を取得
syor_date = batch_context.syor_date
# `emp_chg_inst_lau`をTruncate
await _truncate_emp_chg_inst_lau(db)
_truncate_emp_chg_inst_lau(db)
# emp_chg_inst から、`emp_chg_inst_lau`へInsert
await _insert_into_emp_chg_inst_lau_from_emp_chg_inst(db)
_insert_into_emp_chg_inst_lau_from_emp_chg_inst(db)
# vop_hco_merge_vから、emp_chg_inst_lauをUpdate
await _update_emp_chg_inst_lau_from_vop_hco_merge_v(db, syor_date)
_update_emp_chg_inst_lau_from_vop_hco_merge_v(db, syor_date)
# dcf_inst_mergeから、emp_chg_inst_lauをUpdate
await _update_dcf_inst_merge_from_emp_chg_inst_lau(db, syor_date)
_update_dcf_inst_merge_from_emp_chg_inst_lau(db, syor_date)
logger.debug('##########################')
logger.debug('End All Processing PGM.')
except Exception as e:
raise BatchOperationException(e)
finally:
await db.disconnect()
db.disconnect()
async def _truncate_emp_chg_inst_lau(db: AsyncDatabase):
def _truncate_emp_chg_inst_lau(db: Database):
logger.debug("##########################")
try:
await db.execute("TRUNCATE TABLE src05.emp_chg_inst_lau")
db.execute("TRUNCATE TABLE src05.emp_chg_inst_lau")
except Exception as e:
logger.debug("Error! Truncate Table `emp_chg_inst_lau` is Failed!!!")
logger.exception(e)
raise e
logger.debug("Table `emp_chg_inst_lau` was truncated!")
return
async def _insert_into_emp_chg_inst_lau_from_emp_chg_inst(db: AsyncDatabase):
def _insert_into_emp_chg_inst_lau_from_emp_chg_inst(db: Database):
logger.debug("##########################")
try:
elapsed_time = ElapsedTime()
@ -71,21 +70,22 @@ async def _insert_into_emp_chg_inst_lau_from_emp_chg_inst(db: AsyncDatabase):
WHERE
enabled_flg = 'Y'
"""
res = await db.execute(sql)
res = db.execute(sql)
logging_sql(logger, sql)
logger.info(f'Query OK, {res.rowcount} rows affected ({elapsed_time.of})')
except Exception as e:
logger.debug("Error! Insert into `emp_chg_inst_lau` from `emp_chg_inst` was failed!!!")
raise e
logger.debug("Success! Insert into `emp_chg_inst_lau` from `emp_chg_inst` was inserted!")
return
async def _update_emp_chg_inst_lau_from_vop_hco_merge_v(db: AsyncDatabase, syor_date: str):
def _update_emp_chg_inst_lau_from_vop_hco_merge_v(db: Database, syor_date: str):
# vop_hco_merge_vはデータが作られないため、この洗い替え処理は基本空振りする
logger.debug("##########################")
try:
select_result = await db.execute_select(
select_result = db.execute_select(
"""
SELECT
COUNT(v_inst_cd) AS row_count
@ -106,7 +106,7 @@ async def _update_emp_chg_inst_lau_from_vop_hco_merge_v(db: AsyncDatabase, syor_
logger.info('vop_hco_merge_v Table Data is exists!')
# vop_hco_merge_v から、emp_chg_inst_lauをUpdateします
result = await db.execute_select(
result = db.execute_select(
"""
SELECT
v_inst_cd,
@ -144,16 +144,15 @@ async def _update_emp_chg_inst_lau_from_vop_hco_merge_v(db: AsyncDatabase, syor_
logger.debug(f"emp_chg_inst_lau v_inst_cd could not set from {v_inst_cd_merge} to {v_inst_cd_merge}!")
raise e
logger.debug(f"Success! emp_chg_inst_lau v_inst_cd was set from {v_inst_cd} to {v_inst_cd_merge}!")
return
async def _update_dcf_inst_merge_from_emp_chg_inst_lau(db: AsyncDatabase, syor_date: str):
def _update_dcf_inst_merge_from_emp_chg_inst_lau(db: Database, syor_date: str):
# dcf_inst_mergeから、emp_chg_inst_lauをUpdate
# Get count from DCF_INST_MERGE
logger.debug("##########################")
try:
select_result = await db.execute_select(
select_result = db.execute_select(
"""
SELECT
COUNT(dcf_inst_cd) AS row_count
@ -167,7 +166,6 @@ async def _update_dcf_inst_merge_from_emp_chg_inst_lau(db: AsyncDatabase, syor_d
""",
{'syor_date': syor_date}
)
except Exception as e:
logger.debug("Error! Getting Count of dcf_inst_merge was failed!")
raise e
@ -205,7 +203,7 @@ async def _update_dcf_inst_merge_from_emp_chg_inst_lau(db: AsyncDatabase, syor_d
WHERE
el.inst_cd = dm.dcf_inst_cd
"""
res = await db.execute(
res = db.execute(
update_sql,
{'syor_date': syor_date}
)
@ -216,4 +214,5 @@ async def _update_dcf_inst_merge_from_emp_chg_inst_lau(db: AsyncDatabase, syor_d
raise e
logger.debug("emp_chg_inst_lau.v_inst_cd was set!")
return

48
ecs/jskult-batch-daily/src/db/async_database.py
View File

@ -1,48 +0,0 @@
import asyncio
from sqlalchemy import CursorResult
from tenacity import retry, stop_after_attempt, wait_exponential
from src.db.database import Database
from src.logging.get_logger import get_logger
from src.system_var import environment
logger = get_logger(__name__)
class AsyncDatabase(Database):
"""データベース非同期操作クラス"""
def __init__(self, username: str, password: str, host: str, port: int, schema: str) -> None:
super().__init__(username, password, host, port, schema)
@classmethod
def get_instance(cls):
return cls(
username=environment.DB_USERNAME,
password=environment.DB_PASSWORD,
host=environment.DB_HOST,
port=environment.DB_PORT,
schema=environment.DB_SCHEMA
)
@retry(
wait=wait_exponential(
multiplier=environment.DB_CONNECTION_RETRY_INTERVAL_INIT,
min=environment.DB_CONNECTION_RETRY_INTERVAL_MIN_SECONDS,
max=environment.DB_CONNECTION_RETRY_INTERVAL_MAX_SECONDS
),
stop=stop_after_attempt(environment.DB_CONNECTION_MAX_RETRY_ATTEMPT))
async def connect(self):
await asyncio.get_event_loop().run_in_executor(None, super().connect)
async def execute_select(self, select_query: str, parameters=None) -> list[dict]:
res = await asyncio.get_event_loop().run_in_executor(None, super().execute_select, select_query, parameters)
return res
async def execute(self, query: str, parameters=None) -> CursorResult:
res = await asyncio.get_event_loop().run_in_executor(None, super().execute, query, parameters)
return res
async def disconnect(self):
await asyncio.get_event_loop().run_in_executor(None, super().disconnect)