postgreのスキーマ変更用コード

change_schema_postgre2mysql.py

import os



mysql_dbname = "django"# dockerで決めた値


def removePublicStr(line):
    # postgres専用のスキーマ名を削除
    return line.replace("public.", f"${mysql_dbname}.")

def isSetLine(line):
    new_line = line.lstrip()
    return len(new_line)>3 and new_line[0:3] == "SET"

def isAlterTable(line):
    new_line = line.lstrip()
    alter_str = "ALTER TABLE ONLY"
    return len(new_line)>len(alter_str) and new_line[0:len(alter_str)] == alter_str
def changeAlterTable(line):
    return line.replace("ALTER TABLE ONLY", "ALTER TABLE")


def isCreateTableStart(line):
    new_line = line.lstrip()
    table_str = "CREATE TABLE"
    return len(new_line)>len(table_str) and new_line[0:len(table_str)] == table_str

def isCreateSequenceStart(line):
    new_line = line.lstrip()
    seq_str = "CREATE SEQUENCE"
    return len(new_line)>len(seq_str) and new_line[0:len(seq_str)] == seq_str
def isAlterSequenceStart(line):
    new_line = line.lstrip()
    seq_str = "ALTER SEQUENCE"
    return len(new_line)>len(seq_str) and new_line[0:len(seq_str)] == seq_str

def isTableId(line):
    new_line = line.lstrip()
    return len(new_line)>3 and new_line[0:3] == "id "
def changeTableId(line):
    not_null_str = "NOT NULL,"
    #print(line.rstrip()[-len(not_null_str):], not_null_str, line.rstrip()[-len(not_null_str):] == not_null_str)
    # 既に行われている場合は処理しない
    if line.rstrip()[-len(not_null_str):] == not_null_str:
        #print(line.replace(not_null_str, "NOT NULL AUTO_INCREMENT,"))
        return line.replace(not_null_str, "NOT NULL PRIMARY KEY AUTO_INCREMENT,") # PRIMARY KEY は 今回Alterで処理してたので無視してOK？(必要らしい)
    
    return line

def changeWithOrWithoutTimeTable(line):
    with_time_str = "timestamp with time zone"
    without_time_str = "timestamp without time zone"

    def setDefaultTime(line):
        if "NOT NULL" in line:
            return "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
        else:
            ## NULLを明示しないとダメらしい
            return "TIMESTAMP NULL DEFAULT NULL"

    if with_time_str in line:
        return line.replace(with_time_str, setDefaultTime(line))#デフォルト値がないとエラー？
    elif without_time_str in line:
        return line.replace(without_time_str, setDefaultTime(line))

    return line
def changeCharacterVaryingTable(line):
    ## 変更しなくてもOK?
    chara_varying = "character varying"

    if chara_varying in line:
        return line.replace(chara_varying, "VARCHAR")
    return line
def changeTableData(line):
    return changeWithOrWithoutTimeTable(
        line
        #changeCharacterVaryingTable(line)
    )


def isOperationEnd(line):
    new_line = line.rstrip()
    return len(new_line)>1 and new_line[-1] == ";" 
def changeTableCharaset(line):
    return line.replace(";", "CHARACTER SET utf8;")



def loadFile(path, write_line):
    with open(path) as f:
        in_create_table = False
        in_create_sequence = False

        for s_line in f:
            s_line = removePublicStr(s_line)

            ## 命令始まりをチェック
            if in_create_table or in_create_sequence:
                ...

            elif isSetLine(s_line):
                write_line.append("-- " + s_line)

            elif isAlterTable(s_line):
                write_line.append(changeAlterTable(s_line))

            elif isCreateTableStart(s_line):
                in_create_table = True
            elif isCreateSequenceStart(s_line) or isAlterSequenceStart(s_line):
                in_create_sequence = True

            else:
                write_line.append(s_line)


            # 命令内部の処理（sequenceはコメントアウト）
            if in_create_sequence:
                write_line.append("-- " + s_line)

            elif in_create_table:
                if isTableId(s_line):
                    write_line.append(changeTableId(s_line))

                else:
                    write_line.append(changeTableData(s_line))


            # mysqlの命令文終了かを判断し、内部用boolを非活性
            if isOperationEnd(s_line):
                in_create_sequence = False
                in_create_table = False


def saveFile(path, write_line):
    basename = os.path.basename(path)
    dirname = os.path.dirname(path)
    #print(write_line[10], dirname + basename)
    with open(dirname + "/changed_" + basename, mode="w") as f:
        f.writelines(write_line)


def main():
    while True:
        print("ファイル相対パスを入力↓")
        path = input()
        if os.path.isfile(path):
            break
            
        print("パスが存在しません\n")

    write_line = []
    loadFile(path, write_line)
    saveFile(path, write_line)
    


if __name__ == "__main__":
    main()