0

Uvicorn 不会在线程内运行，因为信号在线程中不起作用。只需删除信号处理即可阻止服务器关闭（需要强制关闭）

我的解决方案是干扰__new__函数以获取服务器对象并创建关闭函数，然后将其绑定到线程外的信号。

然而，这是一个非常丑陋的解决方案。有没有更好的方法？

def run():
    '''
    Start uvicorn server
    returns exit function
    '''
    server = None

    old_new = uvicorn.Server.__new__

    def spoof_server(self, *_, **__):
        '''Interfeer with __new__ to set server'''
        nonlocal server
        server = old_new(self)
        return server

    uvicorn.Server.__new__ = spoof_server
    uvicorn.Server.install_signal_handlers = lambda *_, **__: None

    Thread(target=uvicorn.run, args=[make_app()]).start()

    def exit_server():
        print('exiting...')
        server.handle_exit(None, None)

    return exit_server
4

1 回答 1

1

我也在寻找这样的东西。我发现这个答案对我有帮助。 https://stackoverflow.com/a/64521239/13029591

我将在此处发布片段：

import contextlib
import time
import threading
import uvicorn

class Server(uvicorn.Server):
    def install_signal_handlers(self):
        pass

    @contextlib.contextmanager
    def run_in_thread(self):
        thread = threading.Thread(target=self.run)
        thread.start()
        try:
            while not self.started:
                time.sleep(1e-3)
            yield
        finally:
            self.should_exit = True
            thread.join()

config = Config("example:app", host="127.0.0.1", port=5000, log_level="info")
server = Server(config=config)

with server.run_in_thread():
    # Server is started.
    ...
    # Server will be stopped once code put here is completed
    ...

# Server stopped.
于 2021-03-11T19:53:03.523 回答