标签: quart

我们已经部署了一些 API（很少使用 Node.js/Express.js 开发，其他使用 Python Quart 开发）。我们的所有 API 均使用 Azure 容器化实例进行部署。我们通过 Postman 设置了定期 API 监控。API 在大约 20% 的情况下会失败，并出现错误：套接字挂起。我们在开发区域或通过浏览器访问 API 时从未遇到过此问题。什么可能导致此套接字挂起问题以及我们如何克服它？

我们的 Node.js API Dockerfile 设置如下：

FROM node:16

WORKDIR /app

COPY package*.json ./
RUN npm install
COPY . .
RUN rm -rf .env
RUN mv production.env .env

#ENV PORT=5000
EXPOSE 5000

CMD ["npm", "run", "prod"]

Python Quart API Dockerfile 设置如下：

FROM continuumio/miniconda3

COPY . /api/

WORKDIR /api/src

RUN conda env create -f /api/environment.yml
COPY entrypoint.sh ./
RUN chmod +x ./entrypoint.sh

ENTRYPOINT …

我需要 Python 网络框架工作Quart 的帮助，更具体地说是 websockets。我希望能够在连接时注册客户端（将其添加到 python 列表），并在断开连接时取消注册（从 python 列表中删除它）。我在网上能找到的最接近的是这段代码：

connected = set()

async def handler(websocket, path):
    global connected
    # Register.
    connected.add(websocket)
    try:
        # Implement logic here.
        await asyncio.wait([ws.send("Hello!") for ws in connected])
        await asyncio.sleep(10)
    finally:
        # Unregister.
        connected.remove(websocket)

来源

但这不适用于夸脱 websockets。

帮助将不胜感激。

如果您检查Quart库，app.run()只需建立一些配置，然后使用asyncio.run(serve(self, config))，其中 serve 来自from hypercorn.asyncio import serve。

因此，即使您通过运行Quart应用程序python myapp.py，它不是已经在使用Hypercorn服务器了吗？

特别是，this 和 running via 有hypercorn myapp:app什么区别？

https://pgjones.gitlab.io/quart/deployment.html

不建议在生产中直接运行 Quart（通过 run()）。相反，建议使用 Hypercorn 或替代 ASGI 服务器运行 Quart。Hypercorn 与 Quart 一起安装，默认情况下用于处理请求（例如使用 run()）。

所以听起来，即使Hypercorn默认使用 来处理请求run()，也不建议使用run()? 还有人糊涂吗？

我正在寻找在 Quart 上使用基本身份验证。我知道 quart-auth 可用，但它仅支持基于 cookie 的身份验证。有没有一种方法可以使用基本身份验证，而无需使用 Flask-BasicAuth 的 Flask 补丁？

我正在开发一个 Python 应用程序，但我正在从 Flask 迁移到 Quart。应用程序需要一个在应用程序运行时不断运行的后台任务。

当我尝试使用 control-c 停止进程时，线程没有完全关闭，而是位于关闭例程中的 while 循环中。

    while not self._master_thread_class.shutdown_completed:
        if not pro:
            print('[DEBUG] Thread is not complete')
            pro = True

我已经关注了这个Stackoverflow 问题，但我不知道如何干净地关闭后台线程，所以我希望得到一个解释，因为 Quart 文档似乎有点缺乏。

主线程类：

import asyncio

class MasterThread:

    def __init__(self, shutdown_requested_event):
        self._shutdown_completed = False
        self._shutdown_requested_event = shutdown_requested_event
        self._shutdown_requested = False

    def __del__(self):
        print('Thread was deleted')

    def run(self, loop) -> None:
        asyncio.set_event_loop(loop)
        loop.run_until_complete(self._async_entrypoint())

    @asyncio.coroutine
    def _async_entrypoint(self) -> None:
        while not self. _shutdown_requested and \
            not self._shutdown_requested_event.isSet():
            #print('_main_loop()')
            pass

            if self._shutdown_requested_event.wait(0.1):
                self. _shutdown_requested = …