Flask-RESTful 和 Kivy

Question

我将 Flask-RESTful 与 Kivy 一起使用。用于资源路由的 Flask-RESTful api 和用于 TFT GUI 的 Kivy。两者都是基于类的：Kivy 的主应用 class Touch(App) 和 restful api 的 class Load(Resource)。

class Load(Resource):
    @auth.login_required
    def post(self):
        data = request.json
        # call handle(data)


class Touch(App):
    api.add_resource(Load, '/load')
    def handle(data):
        # do something with data  

我想使用class Touch(App) 中的方法进行路由。我检查了flask-restful documentation 和this post，但没有找到基于方法的资源的示例。它必须是基于类的资源吗？

Answer 1

Flask-RESTful 和 Kivy 可以作为两个独立的进程一起工作。路由保留在 Flask 进程下，该进程使用 Queue 将请求传递给 Kivy 进程。

脚本： 在此脚本中，FLASK 进程以 json 格式获取“加载”请求并将其传递给 KIVY 进程。该脚本适用于 Python2.7 和 Kivy 1.10.0。

#!/usr/bin/python2.7 python2.7
# -*- coding: utf-8 -*-

# kivy modules first, if not Kivy may cause problems
import kivy
from kivy.app import App
from kivy.lang import Builder
from kivy.uix.label import Label
from kivy.uix.floatlayout import FloatLayout
from kivy.uix.screenmanager import ScreenManager, Screen
from kivy.properties import StringProperty
kivy.require('1.10.0')

# common modules
import sys
import os
import time
import signal
from multiprocessing import Process
from multiprocessing import Queue

# Flask modules
from flask import Flask
from flask import request
from flask_restful import reqparse, abort, Api, Resource

# wsgi (Web Server Gateway Interface) modules
import eventlet
from eventlet import wsgi

# json modules
import json, ast

# async server setup
app = Flask(__name__)
api = Api(app)

# global variables
data_json = None

def start_Flask(q):
    print("Starting Flask...")
    q.put([42, None, 'hello'])      # some random initial value
    wsgi.server(eventlet.listen(('', 5000)), app)   # deploy server

def signal_handler(signal, frame):  # process terminator
    print " CTRL + C detected, exiting ... "
    exit(0)

# api resource classes ###############################################
class Load(Resource):
    def post(self):
        data = request.json
        package = ("Load", data)    # create tuple
        q.put(package)  # put tuple into queue

# setup the Api resource routing here ################################
api.add_resource(Load, '/load')

# kivy gui classes ###################################################
# main screen        
class MainScreen(Screen):
    def __init__(self, **kwargs):
        self.name="MAIN SCREEN"
        super(Screen, self).__init__(**kwargs)

class My_Gui(App):
    MainScreenTitle = "MainScreen title"
    MainScreenLabel = "MainScreen label"
    MessageButtonEnter = "GO"
    MessageButtonExit = "EXIT"

    def cancel(self):
        print "load cancelled by user"

    def exit(self):
        print "exiting..."
        p1.terminate()
        exit(1)

    def enter(self):
        print "getting a queue element"
        try:
            if q.empty() == False:
                package = q.get()   # retrieve package from queue
                print("got the package")
                print(package)
        except:
            print "the queue is empty"  

    def build(self):
        sm = Builder.load_string("""

ScreenManager
    MainScreen:
        size_hint: 1, .7
        auto_dismiss: False
        title: app.MainScreenTitle       
        title_align: "center"

        BoxLayout:
            orientation: "vertical"
            Label:
                text: app.MainScreenLabel
            BoxLayout:
                orientation: "horizontal"
                spacing: 10
                size_hint: 1, .5
                Button:
                    text: app.MessageButtonEnter  # start app
                    on_press:
                        app.enter()
                Button:
                    text: app.MessageButtonExit  # exit app
                    on_press:
                        app.exit()
        """)

        return sm


# main #################################################################
if __name__ == '__main__':

    #CTRL+C signal handler
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    q = Queue()

    global p1
    p1 = Process(target=start_Flask, args=(q,)) # assign Flask to a process
    p1.daemon = True
    p1.start()  #launch Flask as separate process
    My_Gui().run()   # run Kivy app