manifest_keeper.py 文件源码

def execute(self, arguments):
        self._strict_assert(arguments.get('external_request_id'))

        postgres = StoragePool.get_connected_storage('BayesianPostgres')

        try:
            results = postgres.session.query(StackAnalysisRequest)\
                        .filter(StackAnalysisRequest.id == arguments.get('external_request_id'))\
                        .first()
        except SQLAlchemyError:
            postgres.session.rollback()
            raise

        manifests = []
        if results is not None:
            row = results.to_dict()
            request_json = row.get("requestJson", {})
            manifests = request_json.get('manifest', [])

        return {'manifest': manifests}