Использование колбы внутри класса

У меня есть приложение со многими потоками. Одной из них является фляга, которая используется для реализации (подмышечного) 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 внутри класса?

Ответ 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 на лету не рекомендуется, ни потокобезопасное

Ответ 2

Прекрасное решение, именно то, что мне было нужно. У меня есть приложение py3, которому нужен сервис отдыха как встроенный сервис... Я добавил базовый класс для передачи некоторых настроек и нескольких методов... Большое спасибо за это...

Ответ 3

Чтобы завершить ответ Костаса Пелелиса, мне было трудно найти причину, по которой в Отклике не использовалось возвращаемое значение Действие.

Вот еще одна версия класса 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