postgresqlconn.py 文件源码

def execute_read_query(self, query: str, args: tuple = ()) -> (list, list):
        """ results are returned as a list of dicts"""
        if not self.__conn:
            raise psycopg2.InterfaceError("null connection")

        rows = None

        try:
            with self.__conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as curs:
                curs.execute(query, args)
                rows = curs.fetchall()
        except Exception as e:
            tb.print_exc()
            raise(e)

        if len(rows) == 0:
            return []
        return rows