Uvicorn/FastAPI 作为带有 cx_Freeze 的 Windows 服务，无法停止服务

Question

我对编程和 Python 很陌生，我使用 FastAPI 创建了一个 API，我需要在 Windows 上运行，我想将它作为服务运行。服务器将启动并正常运行，但是当我尝试停止它时，Windows 会抛出一个错误，我认为这是因为我没有“停止”uvicorn，我不知道该怎么做。我一直在谷歌搜索并尝试在守护线程中运行 Uvicorn，因为我读到当主线程退出时守护线程将关闭，但这也无济于事。 我使用 cx_Freeze 将应用程序转换为 exe。谁能指出我正确的方向？

我正在使用此模板进行 cx_Freeze：https://github.com/marcelotduarte/cx_Freeze/tree/main/cx_Freeze/samples/service

我在 Windows 10、Python 3.9.1、Uvicorn 0.13.3、cx_Freeze 6.5 上

ServiceHandler.py

import os
import sys
import cx_Logging
import api
import logging
import threading
from uvicorn import Config, Server


logging.basicConfig(
    filename = os.path.join(os.path.dirname(sys.executable), "log.txt"),
    level = logging.DEBUG, 
    format = '[API] %(levelname)-7.7s %(message)s'
)


class Handler:

    # no parameters are permitted; all configuration should be placed in the
    # configuration file and handled in the initialize() method
    def __init__(self):
        self.stopEvent = threading.Event()
        self.stopRequestedEvent = threading.Event()

    # called when the service is starting
    def initialize(self, configFileName):
        self.directory = os.path.dirname(sys.executable)
        cx_Logging.StartLogging(os.path.join(self.directory, "testing.log"), cx_Logging.DEBUG)
        #pass

    # called when the service is starting immediately after initialize()
    # use this to perform the work of the service; don't forget to set or check
    # for the stop event or the service GUI will not respond to requests to
    # stop the service
    def run(self):
        cx_Logging.Debug("stdout=%r", sys.stdout)
        sys.stdout = open(os.path.join(self.directory, "stdout.log"), "a")
        sys.stderr = open(os.path.join(self.directory, "stderr.log"), "a")
        self.main()
        self.stopRequestedEvent.wait()
        self.stopEvent.set()


    # called when the service is being stopped by the service manager GUI
    def stop(self):
        try:
            logging.debug("Stopping Service")
            self.stopRequestedEvent.set()
            self.stopEvent.wait()
            # How to stop the server???
        except Exception as e:
            logging.error(e)

    def main(self):
        try:
            self.config = Config(app=api.app, host="0.0.0.0", port=8004, reload=False)
            self.app_server = Server(self.config)
            self.app_server.install_signal_handlers = lambda: None # Need this line, or the server wont start
            self.app_server.run()
        except Exception as e:
            logging.error(e)

setup.py

from cx_Freeze import setup, Executable

options = {
    "build_exe": {
        "packages": ["uvicorn", "fastapi", "pydantic", "threading"],
        "includes": ["ServiceHandler", "cx_Logging", "ipaddress", "colorsys"],
        "excludes": ["tkinter"],
    }
}

executables = [
    Executable(
        "Config.py",
        base="Win32Service",
        target_name="api.exe",
    )
]

setup(
    name="TestService",
    version="0.1",
    description="Sample Windows serice",
    executables=executables,
    options=options,
)

我也读过这个； https://github.com/encode/uvicorn/issues/742 但是我的知识有限，所以我不太明白如何在我的应用程序中实现它？

Answer 1

所以我得到了服务。 Uvicorn/FastAPI 现在将作为 Windows 服务启动和停止，我不知道是否有任何潜在的缺点，但这是工作代码；

import os
import sys
import cx_Logging
import api
import logging
import threading
from uvicorn import Config, Server


logging.basicConfig(
    filename = os.path.join(os.path.dirname(sys.executable), "log.txt"),
    level = logging.DEBUG, 
    format = '[API] %(levelname)-7.7s %(message)s'
)

#My class for creating and running Uvicorn in a thread
class AppServer:
    def __init__(self, app, host: str = "0.0.0.0", port: int = 8004, reload: bool = False):
        self.config = Config(app=app, host=host, port=port, reload=reload)
        self.server = Server(self.config)
        self.server.install_signal_handlers = lambda: None # Need this line, or the server wont start
        self.proc = None
    
    def run(self):
        self.server.run()

    def start(self):
        self.proc = threading.Thread(target=self.run, name="Test", args=())
        self.proc.setDaemon(True)
        self.proc.start()

    def stop(self):
        if self.proc:
            self.proc.join(0.25)


class Handler:

    # no parameters are permitted; all configuration should be placed in the
    # configuration file and handled in the initialize() method
    def __init__(self):
        self.stopEvent = threading.Event()
        self.stopRequestedEvent = threading.Event()

    # called when the service is starting
    def initialize(self, configFileName):
        self.directory = os.path.dirname(sys.executable)
        cx_Logging.StartLogging(os.path.join(self.directory, "testing.log"), cx_Logging.DEBUG)
        #pass

    # called when the service is starting immediately after initialize()
    # use this to perform the work of the service; don't forget to set or check
    # for the stop event or the service GUI will not respond to requests to
    # stop the service
    def run(self):
        cx_Logging.Debug("stdout=%r", sys.stdout)
        sys.stdout = open(os.path.join(self.directory, "stdout.log"), "a")
        sys.stderr = open(os.path.join(self.directory, "stderr.log"), "a")
        self.main()
        self.stopRequestedEvent.wait()
        self.stopEvent.set()


    # called when the service is being stopped by the service manager GUI
    def stop(self):
        try:
            logging.debug("Stopping Service")
            self.server.stop()
            self.stopRequestedEvent.set()
            self.stopEvent.wait()
            # How to stop the server???
        except Exception as e:
            logging.error(e)


    def main(self):
        try:
            logging.debug("Starting server,,,")
            self.server = AppServer(app=api.app)
            self.server.start()
        except Exception as e:
            logging.error(e)

我在这里得到了答案； How to use FastAPI and uvicorn.run without blocking the thread?