diff --git a/ecs/jskult-batch-laundering/src/db/database.py b/ecs/jskult-batch-laundering/src/db/database.py index d2171fb5..280a9d8b 100644 --- a/ecs/jskult-batch-laundering/src/db/database.py +++ b/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) diff --git a/rds_mysql/stored_procedure/src05/whs_org_laundering.sql b/rds_mysql/stored_procedure/src05/whs_org_laundering.sql index aad11100..405f0621 100644 --- a/rds_mysql/stored_procedure/src05/whs_org_laundering.sql +++ b/rds_mysql/stored_procedure/src05/whs_org_laundering.sql @@ -42,6 +42,7 @@ BEGIN '【洗替】1.卸組織洗替② 開始' ); + SET SESSION optimizer_switch = 'derived_merge=off'; INSERT INTO internal05.whs_customer_org_t ( whs_cd,