db.py 文件源码

def get_result(self, perma_id: int) -> Optional[Permadata]:
        try:
            self._health_check()
            with self.conn.cursor() as curs:
                logger.info("retrieving perma_id %s from database", perma_id)
                curs.execute(RETRIEVE_SQL, (perma_id,))
                row = curs.fetchone()

            # If there's no result, return None.
            if row is None:
                return None

            # Otherwise, return a ``Permadata`` instance.
            model_name, request_data, response_data = row
            return Permadata(model_name, json.loads(request_data), json.loads(response_data))
        except (psycopg2.Error, AttributeError):
            logger.exception("Unable to retrieve result")
            return None