Flask 应用程序：处理触发器请求（无服务器，非阻塞）
Flask app: handle requests on trigger (no server, non blocking)
TLDR
考虑一个烧瓶应用程序：
# server.py
import flask
app = flask.Flask(__name__)
# routes, views, etc go here...
# ...
# run app
if __name__ == '__main__':
app.run()
而不是运行连接服务器，我想在我决定时触发请求并获得响应：
# server-trigger.py
from server import app
dummy_request = """GET /123 HTTP/1.1
Host: 127.0.0.1:8888
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
sec-ch-ua: " Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.82 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Accept-Language: he-IL,he;q=0.9,en-US;q=0.8,en;q=0.7"""
res = app.handle_request(dummy_request) # handle request is not real, but needed.
# app.dispatch_request({}) # tried this, doesn't work.
# app.full_dispatch_request() # doesn't work either.
可以使用烧瓶吗？
场景详情
没必要看懂，但有帮助
我已经构建并测试了 Flask 应用程序，目前通过 app.run() 选项在本地提供服务。
而不是 运行在本地安装 Flask 开发服务器，我希望另一个企业服务器来处理对我的应用程序的请求。我正在尝试与该服务器集成。此服务器允许在自定义端点上自定义处理请求。
例如
enterprise-server.com/ - server home endpoint
enterprise-server.com/custom-endpoint - my custom endpoint on this server
我可以 运行 任何 python 我想要的代码，但我必须处理请求和 return 有效响应。意味着这个函数不能启动服务器，因为它会阻塞线程并等待请求，return没有任何响应。
事实是，我已经有了一个我想利用的 Flask 应用程序，而不是重写我的 Flask 应用程序。
所以我想，我的 app 实例（这是 Flask 服务器的一个实例）中已经定义了所有路由，但我不想 运行 服务器并听取请求，我只想在我决定时将请求转发到应用程序实例。
# flask-server.py
# instead of
app.run()
# enterprise-server-custom-endpoint-handle.py
# run this instead:
def handle_custom_endpoint(req:string):
res = app.handle_request(req)
return res
当然，烧瓶路线会进行一些调整，因为它 运行 在其他路径上。
编辑：我重写了我的答案，因为我没有考虑请求的格式（原始）。
我找到了两个选项。选项 1 需要解析 http 请求（使用 http-request-translator) before passing it to the flask test client. Option 2 does the same thing in a much cleaner way making use of low-level Werkzeug modules. Unfortunately, Option 2 doesn't work because of unmaintained code. I've included it anyway, in case someone can maintain/fork the project werkzeug-raw.
app.py:
from flask import Flask
no_server_app = Flask(__name__)
@no_server_app.route('/custom-endpoint/<int:id>')
def test_endpoint(id: int):
return {"input":id,"output":id+1}
if __name__ == '__main__':
no_server_app.run()
endpoint_handler.py（选项 1）：
from flask.testing import FlaskClient
from hrt.interface import HttpRequestTranslator as hrt
from app import no_server_app
class RawClient(FlaskClient):
def open(self,raw_req):
req = hrt(request=raw_req)
req = {'headers':req.headers, **req.details}
# remove unnecessary items from req dict
for k in ['Host','protocol','version','pre_scheme']:
req.pop(k)
return super(RawClient, self).open(**req)
no_server_app.test_client_class = RawClient
client = no_server_app.test_client()
if __name__ == '__main__':
raw_req = 'GET /custom-endpoint/100 HTTP/1.1'
response = client.open(raw_req)
print(response.status_code, response.data)
endpoint_handler.py（选项 2）：
import werkzeug_raw
from flask.testing import FlaskClient
from app import no_server_app
client = no_server_app.test_client()
if __name__ == '__main__':
raw_req = b'GET /custom-endpoint/100 HTTP/1.1'
response = werkzeug_raw.open(client,raw_req)
print(response.status_code, response.data)
根据 Flask/werkzeug 的需要将您的原始请求解析为对象实际上并不难，因此您可以使用一些自己的代码来做到这一点：
def parse_raw_request(raw_request)
lines = raw_request.splitlines()
method, path, server_protocol = lines[0].split(" ")
headers = dict(line.split(":", maxsplit=1) for line in lines[1:] if line)
return method, path, server_protocol, headers
也许在这里和那里有更多的东西可以使它更健壮。然后创建一个 app.test_client 并根据@timothyh 建议的数据发送请求非常简单：
dummy_request = """GET / HTTP/1.1
Host: 127.0.0.1:8888
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
sec-ch-ua: " Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.82 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Accept-Language: he-IL,he;q=0.9,en-US;q=0.8,en;q=0.7"""
method, path, server_protocol, headers = parse_raw_request(dummy_request)
client = app.test_client()
response = client.open(path=path, method=method, headers=headers) # cant set server_protocol
print(response.data)
TLDR
考虑一个烧瓶应用程序：
# server.py
import flask
app = flask.Flask(__name__)
# routes, views, etc go here...
# ...
# run app
if __name__ == '__main__':
app.run()
而不是运行连接服务器，我想在我决定时触发请求并获得响应：
# server-trigger.py
from server import app
dummy_request = """GET /123 HTTP/1.1
Host: 127.0.0.1:8888
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
sec-ch-ua: " Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.82 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Accept-Language: he-IL,he;q=0.9,en-US;q=0.8,en;q=0.7"""
res = app.handle_request(dummy_request) # handle request is not real, but needed.
# app.dispatch_request({}) # tried this, doesn't work.
# app.full_dispatch_request() # doesn't work either.
可以使用烧瓶吗？
场景详情
没必要看懂，但有帮助
我已经构建并测试了 Flask 应用程序，目前通过 app.run() 选项在本地提供服务。
而不是 运行在本地安装 Flask 开发服务器，我希望另一个企业服务器来处理对我的应用程序的请求。我正在尝试与该服务器集成。此服务器允许在自定义端点上自定义处理请求。
例如
enterprise-server.com/ - server home endpoint
enterprise-server.com/custom-endpoint - my custom endpoint on this server
我可以 运行 任何 python 我想要的代码，但我必须处理请求和 return 有效响应。意味着这个函数不能启动服务器，因为它会阻塞线程并等待请求，return没有任何响应。
事实是，我已经有了一个我想利用的 Flask 应用程序，而不是重写我的 Flask 应用程序。
所以我想，我的 app 实例（这是 Flask 服务器的一个实例）中已经定义了所有路由，但我不想 运行 服务器并听取请求，我只想在我决定时将请求转发到应用程序实例。
# flask-server.py
# instead of
app.run()
# enterprise-server-custom-endpoint-handle.py
# run this instead:
def handle_custom_endpoint(req:string):
res = app.handle_request(req)
return res
当然，烧瓶路线会进行一些调整，因为它 运行 在其他路径上。
编辑：我重写了我的答案，因为我没有考虑请求的格式（原始）。
我找到了两个选项。选项 1 需要解析 http 请求（使用 http-request-translator) before passing it to the flask test client. Option 2 does the same thing in a much cleaner way making use of low-level Werkzeug modules. Unfortunately, Option 2 doesn't work because of unmaintained code. I've included it anyway, in case someone can maintain/fork the project werkzeug-raw.
app.py:
from flask import Flask
no_server_app = Flask(__name__)
@no_server_app.route('/custom-endpoint/<int:id>')
def test_endpoint(id: int):
return {"input":id,"output":id+1}
if __name__ == '__main__':
no_server_app.run()
endpoint_handler.py（选项 1）：
from flask.testing import FlaskClient
from hrt.interface import HttpRequestTranslator as hrt
from app import no_server_app
class RawClient(FlaskClient):
def open(self,raw_req):
req = hrt(request=raw_req)
req = {'headers':req.headers, **req.details}
# remove unnecessary items from req dict
for k in ['Host','protocol','version','pre_scheme']:
req.pop(k)
return super(RawClient, self).open(**req)
no_server_app.test_client_class = RawClient
client = no_server_app.test_client()
if __name__ == '__main__':
raw_req = 'GET /custom-endpoint/100 HTTP/1.1'
response = client.open(raw_req)
print(response.status_code, response.data)
endpoint_handler.py（选项 2）：
import werkzeug_raw
from flask.testing import FlaskClient
from app import no_server_app
client = no_server_app.test_client()
if __name__ == '__main__':
raw_req = b'GET /custom-endpoint/100 HTTP/1.1'
response = werkzeug_raw.open(client,raw_req)
print(response.status_code, response.data)
根据 Flask/werkzeug 的需要将您的原始请求解析为对象实际上并不难，因此您可以使用一些自己的代码来做到这一点：
def parse_raw_request(raw_request)
lines = raw_request.splitlines()
method, path, server_protocol = lines[0].split(" ")
headers = dict(line.split(":", maxsplit=1) for line in lines[1:] if line)
return method, path, server_protocol, headers
也许在这里和那里有更多的东西可以使它更健壮。然后创建一个 app.test_client 并根据@timothyh 建议的数据发送请求非常简单：
dummy_request = """GET / HTTP/1.1
Host: 127.0.0.1:8888
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
sec-ch-ua: " Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.82 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Accept-Language: he-IL,he;q=0.9,en-US;q=0.8,en;q=0.7"""
method, path, server_protocol, headers = parse_raw_request(dummy_request)
client = app.test_client()
response = client.open(path=path, method=method, headers=headers) # cant set server_protocol
print(response.data)