在课堂内使用烧瓶

Question

我有许多线程的应用程序.其中一个是烧瓶,用于实施(腋下)API.它使用低负载,从未暴露在互联网上,因此内置烧瓶网络服务器非常好.

我当前的代码如下所示:

class API:
    # ... all other stuff here, skipped
    def run():
        app = flask.Flask('API')

        @app.route('/cmd1')
        def cmd1():
            self.cmd1()

        @app.route('/cmd2')
        def cmd2()
            self.cmd2()

        app.run()

我觉得我做错了,因为所有的文档都说'在模块级创建烧瓶应用'.但我不想这样做 - 它搞乱了我的测试,而API是较大应用程序的一小部分,它有自己的结构和协议(每个'应用程序'是在一个或多个线程中运行的独立类).

如何在课堂内使用Flask？

Answer 1

虽然这有效,但它不符合Flask风格指南.如果需要在项目中包装Flask应用程序,请根据需要创建单独的类并添加应执行的函数

from flask import Flask, Response


class EndpointAction(object):

    def __init__(self, action):
        self.action = action
        self.response = Response(status=200, headers={})

    def __call__(self, *args):
        self.action()
        return self.response


class FlaskAppWrapper(object):
    app = None

    def __init__(self, name):
        self.app = Flask(name)

    def run(self):
        self.app.run()

    def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None):
        self.app.add_url_rule(endpoint, endpoint_name, EndpointAction(handler))


def action():
    # Execute anything

a = FlaskAppWrapper('wrap')
a.add_endpoint(endpoint='/ad', endpoint_name='ad', handler=action)
a.run()

有些事情需要注意:

• EndpointAction应该是一个包装器,它将执行你的函数并生成一个空的200响应.如果您愿意,可以编辑功能
• 端点处理程序可以是__call__定义了方法的任何内容
• 端点名称应该是唯一的,因为它表示视图名称
• 无法在应用程序之后添加端点,因为一旦应用程序启动,线程将阻塞.您可以通过在单独的线程上运行应用程序来启用它,但不建议动态更改URL映射,也不建议线程安全

Answer 2

所以我刚刚遇到了图书馆Flask-Classful

相对来说真的很简单

在类中创建一个简单的 Web 应用程序是这样的：

from flask import Flask
from flask_classful import FlaskView

app = Flask(__name__)

class TestView(FlaskView):

    def index(self):
    # http://localhost:5000/
        return "<h1>This is my indexpage</h1>"

TestView.register(app,route_base = '/')

if __name__ == '__main__':
    app.run(debug=True) 

处理多路由和动态路由也很简单

class TestView(FlaskView):

    def index(self):
    # http://localhost:5000/
        return "<h1>This is my indexpage</h1>"

    def secondpage(self):
    # http://localhost:5000/secondpage
        return "<h1>This is my second</h1>"
    
    def thirdpage(self,name):
    # dynamic route
    # http://localhost:5000/thirdpage/sometext
        return "<h1>This is my third page <br> welcome"+name+"</h1>"

TestView.register(app,route_base = '/')

使用不同的方法添加自己的路线名称也是可能的

from flask_classful import FlaskView,route

class TestView(FlaskView):

    def index(self):
    # http://localhost:5000/
        return "<h1>This is my indexpage</h1>"


    @route('/diffrentname')
    def bsicname(self):
    # customized route
    # http://localhost:5000/diffrentname
        return "<h1>This is my custom route</h1>"
TestView.register(app,route_base = '/')

这提供了为单独的依赖和独立进程创建单独的类和处理程序的可能性，并将它们作为包导入以在主文件或包装文件上运行

from package import Classname
Classname.register(app,route_base = '/')

这是非常简单和面向对象的

Answer 3

完成 Kostas Pelelis 的回答，因为我很难找到 Response 不直接使用 Action 返回值的原因。

这是没有装饰器的另一个版本的 FLASK 类：

class EndpointAction(object):

    def __init__(self, action):
        self.action = action

    def __call__(self, *args):
        # Perform the action
        answer = self.action()
        # Create the answer (bundle it in a correctly formatted HTTP answer)
        self.response = flask.Response(answer, status=200, headers={})
        # Send it
        return self.response

class FlaskAppWrapper(object):

    def add_all_endpoints(self):
        # Add root endpoint
        self.add_endpoint(endpoint="/", endpoint_name="/", handler=self.action)

        # Add action endpoints
        self.add_endpoint(endpoint="/add_X", endpoint_name="/add_X", handler=self.add_X)
        # you can add more ... 

    def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None):
        self.app.add_url_rule(endpoint, endpoint_name, EndpointAction(handler)) 
        # You can also add options here : "... , methods=['POST'], ... "

    # ==================== ------ API Calls ------- ====================
    def action(self):
        # Dummy action
        return "action" # String that will be returned and display on the webpage
        # Test it with curl 127.0.0.1:5000

    def add_X(self):
        # Dummy action
        return "add_X"
        # Test it with curl 127.0.0.1:5000/add_X