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
评论列表
文章目录