SQLの特殊設定が、反映されていなかった、DBの接続部品の横展開漏れが見つかった

ecs/jskult-batch-laundering/src/db/database.py

@@ -1,10 +1,11 @@
 from sqlalchemy import (Connection, CursorResult, Engine, QueuePool,
                         create_engine, text)
 from sqlalchemy.engine.url import URL
+from tenacity import retry, stop_after_attempt, wait_exponential
+
 from src.error.exceptions import DBException
 from src.logging.get_logger import get_logger
 from src.system_var import environment
-from tenacity import retry, stop_after_attempt, wait_exponential
 
 logger = get_logger(__name__)
 
@@ -12,15 +13,17 @@ logger = get_logger(__name__)
 class Database:
     """データベース操作クラス"""
     __connection: Connection = None
-    __engine: Engine = None
+    __transactional_engine: Engine = None
+    __autocommit_engine: Engine = None
     __host: str = None
     __port: str = None
     __username: str = None
     __password: str = None
     __schema: str = None
+    __autocommit: bool = None
     __connection_string: str = None
 
-    def __init__(self, username: str, password: str, host: str, port: int, schema: str) -> None:
+    def __init__(self, username: str, password: str, host: str, port: int, schema: str, autocommit: bool = False) -> None:
         """このクラスの新たなインスタンスを初期化します
 
         Args:
@@ -29,12 +32,14 @@ class Database:
             host (str): DBホスト名
             port (int): DBポート
             schema (str): DBスキーマ名
+            autocommit(bool): 自動コミットモードで接続するかどうか(Trueの場合、トランザクションの有無に限らず即座にコミットされる). Defaults to False.
         """
         self.__username = username
         self.__password = password
         self.__host = host
         self.__port = int(port)
         self.__schema = schema
+        self.__autocommit = autocommit
 
         self.__connection_string = URL.create(
             drivername='mysql+pymysql',
@@ -43,19 +48,23 @@ class Database:
             host=self.__host,
             port=self.__port,
             database=self.__schema,
-            query={"charset": "utf8mb4"}
+            query={"charset": "utf8mb4", "local_infile": "1"},
         )
 
-        self.__engine = create_engine(
+        self.__transactional_engine = create_engine(
             self.__connection_string,
             pool_timeout=5,
             poolclass=QueuePool
         )
 
+        self.__autocommit_engine = self.__transactional_engine.execution_options(isolation_level='AUTOCOMMIT')
+
     @classmethod
-    def get_instance(cls):
+    def get_instance(cls, autocommit=False):
         """インスタンスを取得します
 
+        Args:
+            autocommit (bool, optional): 自動コミットモードで接続するかどうか(Trueの場合、トランザクションの有無に限らず即座にコミットされる). Defaults to False.
         Returns:
             Database: DB操作クラスインスタンス
         """
@@ -64,7 +73,8 @@ class Database:
             password=environment.DB_PASSWORD,
             host=environment.DB_HOST,
             port=environment.DB_PORT,
-            schema=environment.DB_SCHEMA
+            schema=environment.DB_SCHEMA,
+            autocommit=autocommit
         )
 
     @retry(
@@ -75,15 +85,18 @@ class Database:
         ),
         stop=stop_after_attempt(environment.DB_CONNECTION_MAX_RETRY_ATTEMPT),
         retry_error_cls=DBException
-        )
+    )
     def connect(self):
         """
-        DBに接続します。接続に失敗した場合、リトライします。
+        DBに接続します。接続に失敗した場合、リトライします。\n
+        インスタンスのautocommitがTrueの場合、自動コミットモードで接続する。（明示的なトランザクションも無視される）
         Raises:
             DBException: 接続失敗
         """
         try:
-            self.__connection = self.__engine.connect()
+            self.__connection = (
+                self.__autocommit_engine.connect() if self.__autocommit is True
+                else self.__transactional_engine.connect())
         except Exception as e:
             raise DBException(e)
 

rds_mysql/stored_procedure/src05/whs_org_laundering.sql

@@ -42,6 +42,7 @@ BEGIN
                             '【洗替】１．卸組織洗替② 開始'
     );
 
+    SET SESSION optimizer_switch = 'derived_merge=off';
     INSERT INTO
         internal05.whs_customer_org_t (
             whs_cd,