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

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