postgresqlconn.py 文件源码

def execute_write_query(self, query: str, args: tuple = ()) -> int:
        """ perform a write query

        query -- the query string with optional %s placeholders
        args -- optional tuple of values for %s placeholders

        returns number of rows modified
        """
        if not self.__conn:
            raise psycopg2.InterfaceError("null connection")

        try:
            with self.__conn.cursor() as curs:
                curs.execute(query, args)
        except Exception as e:
            tb.print_exc()
            raise(e)

        return curs.rowcount