AiDOPWarehouse/doc/plan/sql/uat-data/apply_sql_file.py

#!/usr/bin/env python3
"""Apply one UpdateScripts SQL file against aidopdev."""

from __future__ import annotations

import argparse
import re
from pathlib import Path

import pymysql
from pymysql.constants import CLIENT


ROOT = Path(__file__).resolve().parents[4]
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"


def connect() -> pymysql.Connection:
    raw = CONFIG.read_text(encoding="utf-8-sig")
    value = next(
        item
        for item in re.findall(r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw)
        if "Database=aidopdev" in item
    )
    parts = {
        item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip()
        for item in value.split(";")
        if "=" in item
    }
    return pymysql.connect(
        host=parts["server"],
        port=int(parts["port"]),
        user=parts["uid"],
        password=parts["pwd"],
        database=parts["database"],
        charset="utf8mb4",
        autocommit=True,
        client_flag=CLIENT.MULTI_STATEMENTS,
        cursorclass=pymysql.cursors.DictCursor,
    )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("sql_file")
    args = parser.parse_args()
    path = Path(args.sql_file)
    if not path.is_absolute():
        path = ROOT / path
    sql = path.read_text(encoding="utf-8")
    conn = connect()
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            while True:
                rows = cur.fetchall()
                if rows:
                    print(rows)
                if not cur.nextset():
                    break
    finally:
        conn.close()


if __name__ == "__main__":
    main()