db.py 文件源码

def add_result(self,
                   headers: JsonDict,
                   model_name: str,
                   inputs: JsonDict,
                   outputs: JsonDict) -> Optional[int]:
        try:
            self._health_check()
            with self.conn.cursor() as curs:
                logger.info("inserting into the database")

                curs.execute(INSERT_SQL,
                             {'model_name'   : model_name,
                              'headers'      : json.dumps(headers),
                              'request_data' : json.dumps(inputs),
                              'response_data': json.dumps(outputs),
                              'timestamp'    : datetime.datetime.now()})

                perma_id = curs.fetchone()[0]
                logger.info("received perma_id %s", perma_id)

            return perma_id
        except (psycopg2.Error, AttributeError):
            logger.exception("Unable to insert permadata")
            return None