将Python Flask Web服务转换为Windows服务的实现指导请求

我刚接触Python，写了一个Flask API，原本打算用bat文件配合任务计划程序在服务器上运行，但遇到了文件权限问题——直接运行Python代码时能生成日志文件，用任务计划运行就不行。

所以我想改用Windows服务来部署，这样灵活性更高。下面是我的Windows服务代码和简化版API代码（没法提供完整代码），我的API主要逻辑是：查询数据库获取数据，用文档模板填充占位符，最终把文档保存为PDF到指定文件夹，还有一个Web UI来调用这个API生成文档。

我参考了相关的实现思路，但不知道怎么把Flask API正确集成到Windows服务代码里，已经在service.py里导入了Flask API模块，但不清楚具体怎么调用，希望有人能指导我怎么做。

API Python代码 #

import ast
import logging
import os
import traceback
import pandas as pd
import win32com.client
import pythoncom
from docx.shared import Mm
from docxtpl import DocxTemplate, RichText, InlineImage
from sqlalchemy import create_engine
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import sessionmaker
from datetime import datetime
from flask import Flask, request, jsonify
from waitress import serve

app = Flask(__name__)
wdFormatPDF = 17
logging.basicConfig(filename='Logfile.log', format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
# path setting
base_dir = os.path.dirname(os.path.abspath(__file__))
ET_doc = DocxTemplate(template_Path + "Generated.docx")

def narrative_settings(structure_No, cla_engine):
    # ticket 4912 - Active column included
    ns_select_query = "SELECT * FROM Settings where structureNo = '{structure_number}' and Active = 1".format(structure_number=structure_No)
    settings_df = pd.read_sql_query(ns_select_query, cla_engine)
    image_path_var = settings_df['logoFile'].iloc[0]
    ns_report_location = settings_df['FolderPath'].iloc[0]
    eot_logo = InlineImage(ET_doc, image_descriptor=image_path_var, width=Mm(15), height=Mm(15))
    ds_logo = InlineImage(DS_doc, image_descriptor=image_path_var,
    width=Mm(15), height=Mm(15))
    return eot_logo, ds_logo, ns_report_location

@app.route("/generation/", methods=['GET'])
def base():
    report_date_time = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
    eot_logo_var, ds_logo_var, report_location = narrative_settings(structure_no, cla_engine)
    
    context = {'structure_no': structure_no, 'Logo': eot_logo_var,
    'report_time': report_date_time}
    ET_doc.render(context)
    ET_doc.save(report_location + doc_file_name)
    pdf_file_name = f'Narrative_{subj_id}_{report_date}.pdf'
    word = win32com.client.Dispatch("Word.Application", pythoncom.CoInitialize())
    doc = word.Documents.Open(report_location + doc_file_name)
    doc.SaveAs(report_location + pdf_file_name, FileFormat=wdFormatPDF)
    doc.Close()
    word.Quit()
    os.remove(report_location + doc_file_name)
    report_status = "Generated"

if __name__ == "__main__":
     serve(app, host="", port=7105)
     app.run(debug=True)

Service.py代码 #

import servicemanager
import sys
import win32serviceutil
import threading
import concurrent.futures
import time

from na_webservice import app

class workingthread(threading.Thread):
    def __init__(self, quitEvent):
        self.quitEvent = quitEvent
        self.waitTime = 1
        threading.Thread.__init__(self)

    def run(self):
        try:
            # Running start_flask() function on different thread, so that it doesn't blocks the code
            executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)
            executor.submit(self.start_flask)
        except:
            pass

        # Following Lines are written so that, the program doesn't get quit
        # Will Run a Endless While Loop till Stop signal is not received from Windows Service API
        while not self.quitEvent.isSet():  # If stop signal is triggered, exit
            time.sleep(1)

    def start_flask(self):
        # This Function contains the actual logic, of windows service
        # This is case, we are running our flaskserver
        test = FlaskServer()
        test.start()

class FlaskService(win32serviceutil.ServiceFramework):
    _svc_name_ = "Service"
    _svc_display_name_ = "Narrative Service"
    _svc_description_ = "Web Service for executing python code"

    def __init__(self, args):
        win32serviceutil.ServiceFramework.__init__(self, args)
        self.hWaitStop = threading.Event()
        self.thread = workingthread(self.hWaitStop)

    def SvcStop(self):
        self.hWaitStop.set()

    def SvcDoRun(self):
        thread = Thread(target=app.run())
        thread.daemon = True
        thread.start()
        while (1):
            rc = win32event.WaitForSingleObject(self.hWaitStop, 1000)
            if rc == win32event.WAIT_OBJECT_0:
                # Stop event
                break
        self.thread.start()
        self.hWaitStop.wait()
        self.thread.join()
        self.start(app);


if __name__ == '__main__':
    if len(sys.argv) == 1:
        servicemanager.Initialize()
        servicemanager.PrepareToHostSingle(FlaskService)
        servicemanager.StartServiceCtrlDispatcher()
    else:
        win32serviceutil.HandleCommandLine(FlaskService)

解决方案指导 #

你的Service代码里有几个需要修正的地方，下面是逐步调整的方法：

1. 修复Flask运行逻辑
   • 你在workingthread的start_flask方法里提到的FlaskServer类并没有定义，直接用你导入的app即可。
   • 不要用app.run()（它是阻塞的），而是沿用你API里的waitress.serve来启动服务，waitress更适合生产环境，也能更好地配合Windows服务的线程模型。

修改后的workingthread类的start_flask方法：

def start_flask(self):
    # 和API里的启动逻辑保持一致，用waitress运行服务
    serve(app, host="0.0.0.0", port=7105)

2. 修正SvcDoRun方法的冗余代码
   原来的SvcDoRun里有未导入就使用的Thread、错误的self.start(app)调用等问题，简化后如下：

def SvcDoRun(self):
    # 启动工作线程
    self.thread.start()
    # 等待服务停止信号
    self.hWaitStop.wait()
    # 等待线程安全结束
    self.thread.join()

3. 解决文件路径与权限问题

   • 日志、模板、PDF保存路径都要改用绝对路径，避免Windows服务运行时找不到文件，比如API里的日志配置可以改成：

     log_path = os.path.join(base_dir, 'Logfile.log')
     logging.basicConfig(filename=log_path, format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
     

   • Windows服务默认以Local System账户运行，要确保这个账户有访问数据库、模板文件夹、PDF保存目录的读写权限，或者在服务属性里配置为以拥有对应权限的用户身份运行。

4. 安装与测试服务

   • 打开管理员权限的命令提示符，运行以下命令安装服务：

     python Service.py install
     

   • 启动服务：

     python Service.py start
     

   • 调试阶段可以先运行python Service.py debug，不用安装成服务就能直接查看运行状态和报错信息。

备注：内容来源于stack exchange，提问作者Karthik Venkatraman