Приложение Flask для доступа к маршрутам, определенным в классе

Question

Я новичок в Python, у меня есть класс, и на нем определены два маршрута. Как мне привязать эти маршруты к основному работающему серверу.

Вот мой restinput.py

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import logging

from rasa_core.channels import HttpInputChannel
from rasa_core import utils
from rasa_core.agent import Agent
from rasa_core.interpreter import RasaNLUInterpreter
from rasa_core.channels.channel import UserMessage
from rasa_core.channels.direct import CollectingOutputChannel

from flask import Blueprint, request, jsonify

logger = logging.getLogger(__name__)

class RestInput(InputChannel):

    @classmethod
    def name(cls):
        return "rest"

    @staticmethod
    def on_message_wrapper(on_new_message, text, queue, sender_id):
        collector = QueueOutputChannel(queue)

        message = UserMessage(text, collector, sender_id,
                              input_channel=RestInput.name())
        on_new_message(message)

        queue.put("DONE")

    def _extract_sender(self, req):
        return req.json.get("sender", None)


    def _extract_message(self, req):
        return req.json.get("message", None)

    def stream_response(self, on_new_message, text, sender_id):
        from multiprocessing import Queue

        q = Queue()

        t = Thread(target=self.on_message_wrapper,
                   args=(on_new_message, text, q, sender_id))
        t.start()
        while True:
            response = q.get()
            if response == "DONE":
                break
            else:
                yield json.dumps(response) + "\n"

    def blueprint(self, on_new_message):
        custom_webhook = Blueprint(
                'custom_webhook_{}'.format(type(self).__name__),
                inspect.getmodule(self).__name__)

        @custom_webhook.route("/", methods=['GET'])
        def health():
            return jsonify({"status": "ok"})

        @custom_webhook.route("/webhook", methods=['POST'])
        def receive():
            sender_id = self._extract_sender(request)
            text = self._extract_message(request)
            should_use_stream = utils.bool_arg("stream", default=False)

            if should_use_stream:
                return Response(
                        self.stream_response(on_new_message, text, sender_id),
                        content_type='text/event-stream')
            else:
                collector = CollectingOutputChannel()
                on_new_message(UserMessage(text, collector, sender_id,
                                           input_channel=self.name()))
                return jsonify(collector.messages)

        return custom_webhook
def run(serve_forever=True):
    interpreter = RasaNLUInterpreter("models/nlu/default/current")
    action_endpoint = EndpointConfig(url="http://localhost:5055/webhook")
    agent = Agent.load("models/dialogue", interpreter=interpreter,action_endpoint=action_endpoint)
    input_channel = RestInput()
    if serve_forever:
        agent.handle_channels([InputChannel(5004, "/chat", input_channel)])
    return agent

Это мой app.py

from flask import Flask
from flask_cors import CORS
from restinput import RestInput
app = Flask(__name__)
CORS(app)

@app.route("/")
def index():
    return "Hello"

@app.route("/parse",methods=['POST'])
def chat():
    #some code to access the class on this route


if __name__ == "__main__":
    app.run(host='0.0.0.0')

Что я должен добавить в маршрут "/parse", чтобы иметь возможность доступа к маршрутам из класса RestInput.

Я попытался создать агента в / parse-маршруте, а затем получить доступ к другим маршрутам, таким как / parse / webhook, но он не работает. Я просмотрел проект колбы и просмотр, но я не могу понять это.

Answer 1

Маршруты в RestInput прикреплены к Blueprint. Чтобы прикрепить эти маршруты к вашему основному приложению, вы должны зарегистрировать план:

# app.py
some_function = ...  # whatever you want
app.register_blueprint(RestInput().blueprint(some_function))