flightinfo_sever/Functions/DButilsFunction/flightDB.py

import psycopg2
import traceback
from unitls.LogerinTxt import app_logger

class flightDB():
    def __init__(self, host, port, user, password, database):

        self.conn = psycopg2.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database
        )
        self.c = self.conn.cursor()

    def initTable(self, tableName:str,primarykey:str,primarykeyStr:str,keyDict:dict):
        try:
            keyStr = ''
            num = 1
            for key in keyDict:
                if num !=len(keyDict):
                    keyStr += '{} {},\n'.format(key,keyDict[key])
                else:
                    keyStr += '{} {}'.format(key, keyDict[key])
                num +=1

            curStr ="""
                            create table if not exists {} (
                            {} {} ,
                            {}
                            )
                            """.format(tableName, primarykey, primarykeyStr,keyStr)
            #print(curStr)
            self.c.execute(curStr)

            self.conn.commit()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())


    def insertData(self, tableName:str, data:dict, *args):
        try:
            #curStr1 = 'insert into {} '.format(tableName)
            num = 1
            curStr2 = ''
            for key in data:
                if num != len(data):
                    curStr2+='{},'.format(key)
                else:
                    curStr2+='{}'.format(key)
                num+=1
            curStr3 = ''
            num = 1
            for key in data:
                if num != len(data):
                    curStr3+="'{}',".format(data[key])
                else:
                    curStr3+="'{}'".format(data[key])
                num+=1
            curStr = """
                    insert into {} ({})
                    values ({})
                    """.format(tableName, curStr2, curStr3)
            #print(curStr)
            self.c.execute(curStr)

            if args==():
                self.conn.commit()
            return "ok"
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())
            return "fail"

    def lazyInsertData(self,tableName:str, data:dict):
        self.insertData(tableName, data, 'lazy')
    def lazydeleteTable(self,tablename:str):
        try:
            curStr = """delete from {}""".format(tablename)
            self.c.execute(curStr)
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
    def lazyInsertData2(self,tableName:str, curStr2, curStr3):
        try:
            curStr = """insert into {} ({})values {}""".format(tableName, curStr2, curStr3)
            self.c.execute(curStr)
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
    def lazyInsertData3(self,curStr1, curStr2, curStr3):
        try:
            curStr = """insert into display  (ID, A, B)values (%s,'%s','%s')"""%(curStr1, curStr2, curStr3)
            self.c.execute(curStr)
        except Exception  as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)

    def FunctionCommit(self):
        self.conn.commit()

    def FunctionRollback(self):
        self.conn.rollback()

    def deleteTable(self, tablename:str, *condition:str):
        try:
            if condition !=():
                curStr = """
                        delete from {} where {}
                        """.format(tablename, condition[0])
            else:
                curStr = """
                        delete from {}
                        """.format(tablename)
            self.c.execute(curStr)
            self.conn.commit()
            return "ok"
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())
            return "fail"


    def copyTable(self, oldtablename:str, newTablename:str):
        try:
            curStr = """
                    drop table {}
                    """.format(newTablename)
            self.c.execute(curStr)
            self.conn.commit()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())

    def sortTable(self, tablename:str, tableKey:str, fn):
        try:
            curStr = """
                    select * from {} order by {} {}
                    """.format(tablename, tableKey, fn)
            #print(curStr)
            self.c.execute(curStr)
            return self.c.fetchall()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())

    def sort_queryTable(self, findkey:str,tablename:str, condition:str,tableKey:str, fn):
        try:
            curStr = """
                    select {} from {} where {} order by {} {}
                    """.format(findkey,tablename, condition,tableKey, fn)
            #print(curStr)
            self.c.execute(curStr)
            return self.c.fetchall()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())
    def sort_queryTable2(self, findkey:str,tablename:str, condition:str,tableKey:str, fn,tableKey1:str, fn1):
        try:
            curStr = """
                    select {} from {} where {} order by {} {},{} {}
                    """.format(findkey,tablename, condition,tableKey, fn,tableKey1, fn1)
            #print(curStr)
            self.c.execute(curStr)
            return self.c.fetchall()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())

    def queryTabel(self, tablename: str, key: str, condition: str):
        try:
            # 检查表是否存在
            curStr1 = """
                SELECT EXISTS (
                  SELECT * FROM pg_catalog.pg_tables
                  WHERE tablename = %s AND schemaname = 'public'
                );
            """
            self.c.execute(curStr1, (tablename.lower(),))
            table_exists = self.c.fetchone()[0]

            if table_exists:
                # 防止SQL注入，使用参数化查询
                curStr = f"SELECT {key} FROM {tablename} WHERE {condition}"
                self.c.execute(curStr)
                return self.c.fetchall()
            else:
                print(f"表 '{tablename}' 不存在")
                return []  # 返回空列表而不是None，便于统一处理
        except Exception as e:
            app_logger.log_error(e)
            print(traceback.format_exc())
            self.conn.rollback()
            return []




    def getAlldata(self, tablename:str):
        try:
            curStr = """select * from {}""".format(tablename)
            self.c.execute(curStr)
            return self.c.fetchall()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())

    def upDateItem(self, tablename:str, dateDic:dict, condition:str, *args):
        try:
            setStr = ''
            for key in dateDic:
                if tablename == 'display':
                    setStr += '{}={},'.format(key, str(dateDic[key]).replace("'","''"))
                else:
                    setStr += '{}={},'.format(key, dateDic[key])
            if tablename == 'display':
                setStr = setStr[:-1].replace('"',"'")
            else:
                setStr = setStr[:-1]
            curStr = """
                    update {} set {} where {}
                    """.format(tablename, setStr, condition)
            #print(curStr)
            if setStr !="":
                self.c.execute(curStr)
            if args == ():
                self.conn.commit()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())


    def lazyUpdateItem(self,tablename:str, dateDic:dict, condition:str):
        self.upDateItem(tablename,dateDic,condition,'lazy')

    def getSingledata(self,findkey:str,tablename:str):
        try:
            curStr = """
                    select {} from {}
                    """.format(findkey,tablename)
            self.c.execute(curStr)
            return self.c.fetchall()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())

    def deleteSingledata(self,tablename:str,findkey:str):
        try:
            curStr = """
                    delete from {} where {}
                    """.format(tablename,findkey)
            self.c.execute(curStr)
            #print(curStr)
            self.conn.commit()
        except Exception as e:
            app_logger.log_error(curStr)
            app_logger.log_error(e)
            print(traceback.format_exc())

    def close(self):
        try:
            self.conn.close()
        except Exception as e:
            app_logger.log_error(e)
            print(traceback.format_exc())