在 Jupyter Notebook 中调试 Flask 服务器

Question

我想在 jupyter notebook 中调试小型烧瓶服务器进行演示。

我在最新的 Ubuntu 和 Python2 上创建了 virtualenv（在使用 Python3 的 Mac 上也会出现这个错误），pip install flask jupyter。

但是，当我使用 helloworld 脚本创建单元格时，它不会在笔记本中运行。

 from flask import Flask
app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello World!"

if __name__ == "__main__":
    app.run(debug=True,port=1234)

文件“/home/***/test/local/lib/python2.7/site-packages/ipykernel/kernelapp.py”，第 177 行，在 _bind_socket s.bind(“tcp://%s:%i” % (self.ip, port)) 文件“zmq/backend/cython/socket.pyx”，第 495 行，在 zmq.backend.cython.socket.Socket.bind (zmq/backend/cython/socket.c:5653)文件“zmq/backend/cython/checkrc.pxd”，第 25 行，在 zmq.backend.cython.checkrc._check_rc (zmq/backend/cython/socket.c:10014) raise ZMQError(errno) ZMQError: Address already in use

注意 - 每次失败后我都会更改端口号。

当然，它作为独立脚本运行。

不更新 (debug=True) 就可以了。

原文由 chro 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

我安装了 Jupyter 和 Flask，您的原始代码可以正常工作。

---

flask.Flask 对象是 WSGI 应用程序，而不是服务器。当您在 shell 中调用 python -m flask run 时，Flask 使用 Werkzeug 的开发服务器作为 WSGI 服务器。它创建一个新的 WSGI 服务器，然后将您的应用程序作为参数传递给 werkzeug.serving.run_simple 。也许您可以尝试手动执行此操作：

 from werkzeug.wrappers import Request, Response
from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello World!"

if __name__ == '__main__':
    from werkzeug.serving import run_simple
    run_simple('localhost', 9000, app)

Flask.run() --- 在内部调用 run_simple() ，所以这里应该没有区别。

原文由 yorodm 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

诀窍是在单独的线程中运行 Flask 服务器。此代码允许注册数据提供者。主要特点是

• 为服务器找到一个空闲端口。如果您在不同的笔记本中运行服务器的多个实例，它们将竞争同一个端口。

• register_data 函数返回服务器的 URL，因此您可以根据需要使用它。

• 服务器按需启动（当第一个数据提供者注册时）

• 注意： 我从 flask-cors 包中添加了 @cross_origin() 装饰器。否则您无法在笔记本中调用 API 表单。

• 注意： 无法在此代码中停止服务器…

• 注意： 代码使用打字和 python 3 。

• 注意： 目前没有很好的错误处理

import socket
import threading
import uuid
from typing import Any, Callable, cast, Optional

from flask import Flask, abort, jsonify
from flask_cors import cross_origin
from werkzeug.serving import run_simple

app = Flask('DataServer')

@app.route('/data/<id>')
@cross_origin()
def data(id: str) -> Any:
    func = _data.get(id)
    if not func:
        abort(400)
    return jsonify(func())

_data = {}

_port: int = 0

def register_data(f: Callable[[], Any], id: Optional[str] = None) -> str:
    """Sets a callback for data and returns a URL"""
    _start_sever()
    id = id or str(uuid.uuid4())
    _data[id] = f
    return f'http://localhost:{_port}/data/{id}'

def _init_port() -> int:
    """Creates a random free port."""
    # see https://stackoverflow.com/a/5089963/2297345
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(('localhost', 0))

    port = sock.getsockname()[1]
    sock.close()
    return cast(int, port)

def _start_sever() -> None:
    """Starts a flask server in the background."""
    global _port
    if _port:
        return
    _port = _init_port()
    thread = threading.Thread(target=lambda: run_simple('localhost', _port, app))
    thread.start()

原文由 Michael_Scharf 发布，翻译遵循 CC BY-SA 4.0 许可协议