- mysql_gtid.py
from __future__ import annotationsimport asyncio
import contextlib
import os
import sys
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanagerfrom tortoise import BaseDBAsyncClient, connections
from tortoise.backends.base.config_generator import expand_db_url
from tortoise.exceptions import OperationalError@asynccontextmanager
async def tortoise_context() -> AsyncGenerator[BaseDBAsyncClient]:password = os.getenv("TORTOISE_MYSQL_PASS", "123456")db_url = f"mysql://root:{password}@127.0.0.1:3306/mysql"connections_config = {"default": expand_db_url(db_url)}await connections._init(connections_config, create_db=False)try:yield connections.get("default")finally:await connections.close_all()class MysqlGtid:def __init__(self, conn: BaseDBAsyncClient) -> None:self.conn = connasync def get_var_value(self, statement: str) -> str:print(f"--> {statement}")result = await self.conn.execute_query_dict(statement)return str(result[0]["Value"])async def get_gtid_consistency(self) -> str:statement = "SHOW VARIABLES LIKE 'enforce_gtid_consistency';"return await self.get_var_value(statement)async def is_enforce_gtid(self) -> bool:return (await self.get_gtid_consistency()).upper() == "ON"async def get_gtid_mode_status(self) -> str:statement = "SHOW VARIABLES LIKE 'gtid_mode';"return await self.get_var_value(statement)async def execute_script(self, statement: str) -> None:print(f"--> {statement}")await self.conn.execute_script(statement)async def set_enforce_gtid_off(self, mode_on: bool, gtid_mode: str) -> None:statement = "SET GLOBAL enforce_gtid_consistency = OFF;"if mode_on:if gtid_mode == "ON":await self.execute_script("SET GLOBAL gtid_mode = ON_PERMISSIVE;")await self.execute_script("SET GLOBAL gtid_mode = OFF_PERMISSIVE;")await self.execute_script("SET GLOBAL gtid_mode = OFF;")await self.execute_script(statement)async def set_enforce_gtid_on(self, mode_on: bool, origin_gtid_mode: str) -> None:statement = "SET GLOBAL enforce_gtid_consistency = ON;"await self.execute_script(statement)if mode_on:current_status = (await self.get_gtid_mode_status()).upper()if current_status == origin_gtid_mode.upper():returnwith contextlib.suppress(OperationalError):if current_status == "OFF":await self.execute_script("SET GLOBAL gtid_mode = OFF_PERMISSIVE;")await self.execute_script("SET GLOBAL gtid_mode = ON_PERMISSIVE;")if origin_gtid_mode.upper() == "ON":await self.execute_script("SET GLOBAL gtid_mode = ON;")async def main() -> None:async with tortoise_context() as conn:db = MysqlGtid(conn)if "--list" in sys.argv:print(await db.get_gtid_consistency())print(await db.get_gtid_mode_status())elif "--on" in sys.argv:if (await db.get_gtid_mode_status()).upper() == "ON":print("gtid_mode is ON, nothing to do.")returnawait db.set_enforce_gtid_on(True, "ON")else:if not await db.is_enforce_gtid():print("enforce_gtid_consistency is OFF, nothing to do.")returnorigin_gtid_mode = await db.get_gtid_mode_status()gtid_mode = origin_gtid_mode.upper()mode_on = gtid_mode.startswith("ON")await db.set_enforce_gtid_off(mode_on, gtid_mode)if __name__ == "__main__":asyncio.run(main())
Usage::
python mysql_gtid.py --list
python mysql_gtid.py --on
python mysql_gtid.py --off
