app.py 文件源码

def get_task_status(self, task_id):
        """"""
        # type: (str)->Union[str,Dict[str,Any]]

        sql_update = "SELECT task_status, description updated_at FROM {} WHERE task_id=%s".format(self.table_task_status)
        cur = self.db_connection.cursor(cursor_factory=RealDictCursor)  # type: cursor

        try:
            cur.execute(sql_update, (task_id, ))
            tasK_status_record = cur.fetchone()
        except:
            traceback_message = traceback.format_exc()
            return traceback_message
        else:
            cur.close()
            return tasK_status_record