极客观点
项目管理
HarmonyOS
问题描述
我有一个小项目，使用python 的flask写的。项目内有一个接口，功能是需要多次查询数据库，取得数据之后再处理，最后生成word文档。整个接口运行时间较长，大概需要30秒。当我访问这个生成文件的接口的同时再访问本项目其他的接口，它就必须要我等待文件生成的接口有了返回之后才处理。
环境
系统 windows server 2012
Python 3.9.1
Flask 2.1.2
pymssql 2.2.5
PyMySQL 1.0.2问题一
我用ORM不习惯，所以我使用pymssql、pymysql封装了一个类，直接通过SQL进行查询取数等操作，我经过网上查询相关信息，怀疑是pymssql、pymysql不支持多线程导致，这是我封装的数据库操作类：
- 请问问题描述的问题是不是因为此SQL取数代码不支持多线程导致的？我可以怎么改进？
import traceback import datetime import datetime from decimal import Decimal import pymssql import pymysql class databases_action: ''' db_name 通过db_config，获取到对应的数据库连接信息，来进行数据库连接 sql 可以是一句SQL，也可以是一个SQL列表 ''' def __init__(self,db_name,sql): self.db_name = db_name self.sql = sql self.db_config = { 'cost': {'serverName': '192.168.1.10', 'userName': 'rdsuser', 'passWord': 'HB@2088**@kg', 'databases': 'NG0001', 'db_class': 'pymssql'}, 'formal': {'serverName': '192.168.1.11', 'userName': 'rdsuser', 'passWord': 'HB@20jt88**@kg', 'databases': 'NG0001', 'db_class': 'pymssql'}, 'finance': {'serverName': '192.168.1.12', 'userName': 'LC0029999', 'passWord': 'Admin123', 'databases': 'cwbase002', 'db_class': 'pymssql'}, 'oa': {'serverName': '192.168.1.13', 'userName': 'ekp', 'passWord': '123123', 'databases': 'ekp_db', 'db_class': 'pymysql'}, 'fined_bi': {'serverName': '192.168.1.14', 'userName': 'root', 'passWord': 'H87tF@nruan#220505', 'databases': 'finedata', 'db_class': 'pymysql'}, 'main_db': {'serverName': '192.168.7.172', 'userName': 'root', 'passWord': 'h87T#zsjPt@220530', 'databases': 'de_prod', 'db_class': 'pymysql'}, 'yunshu': {'serverName': '192.168.7.51', 'userName': 'root', 'passWord': 'h87T#@02he20220923', 'databases': 'cloudpivot', 'db_class': 'pymysql'}, 'i_localhost': {'serverName': '192.168.1.102', 'userName': 'sa', 'passWord': 'PLOat7456#@!', 'databases': 'NG0001', 'db_class': 'pymssql'}, 'oa_localhost': {'serverName': '127.0.0.1', 'userName': 'root', 'passWord': 'PLOat7456#@!', 'databases': 'ekp', 'db_class': 'pymysql'}, 'fileSystem': {'serverName': '192.168.7.88', 'userName': 'isa', 'passWord': 'infosoft4', 'databases': 'zjk', 'db_class': 'pymssql'}, } ''' 查询 ''' def select_sql(self): try: if self.db_name in self.db_config.keys(): if self.db_config[self.db_name]['db_class'] == 'pymssql': db = pymssql.connect(host=self.db_config[self.db_name]['serverName'], user=self.db_config[self.db_name]['userName'], password=self.db_config[self.db_name]['passWord'], database=self.db_config[self.db_name]['databases']) cursor = db.cursor() else: db = pymysql.connect(host=self.db_config[self.db_name]['serverName'], user=self.db_config[self.db_name]['userName'], password=self.db_config[self.db_name]['passWord'], database=self.db_config[self.db_name]['databases']) cursor = db.cursor() else: return_dict = {'return_code': 403, 'return_info': '不存在的连接方式', 'result': 'E', 'error': '不存在的连接方式'} return return_dict except: return_dict = {'return_code': 402, 'return_info': '数据库连接失败，请检查连接IP、账号、密码、网络等是否正确', 'result': 'E', 'error': '数据库连接失败，请检查连接条件是否正确'} return return_dict try: print(self.sql) cursor.execute(self.sql) results = cursor.fetchall() keys = [] col = cursor.description for column in col: keys.append(column[0]) key_number = len(keys) json_data = [] for row in results: item = dict() for q in range(key_number): # 判断这个数值是不是Decimal，如果有，则转成str if isinstance(row[q], Decimal): item[keys[q]] = str(row[q]) # 判断这个数值是不是date日期，如果是，则转成str elif isinstance(row[q], datetime.datetime): item[keys[q]] = str(row[q]) else: item[keys[q]] = row[q] json_data.append(dict(item)) cursor.close() db.close() if json_data: return_dict = {'return_code': 200, 'return_info': '数据获取正常', 'result': 'S', 'data': json_data} return return_dict else: return_dict = {'return_code': 400, 'return_info': 'SQL返回结果为空，请检查SQL条件', 'result': 'E'} return return_dict except Exception: val = traceback.format_exc() return_dict = {'return_code': 401, 'return_info': '数据库查询失败,请检查SQL语句是否正确', 'result': 'E', 'error': val} return return_dict
问题二
flask的部署，是我直接从网上抄过来的一段代码，我不是很清楚它是否真的支持多线程。我使用了接口测试工具apiPost的并发测试，测试了同项目一些其它简单的接口，它显示支持的并发有一秒40多，但是为什么在我访问问题一的接口时，还需要等待运行完。
- 这段flask部署的代码是否可以直接在生成环境使用？为什么？
问题一的问题是否是因为使用这段代码导致的？如果是，那么我最好使用什么方式去部署这个flask项目？
def run(MULTI_PROCESS): if MULTI_PROCESS == False: WSGIServer(('0.0.0.0', 8086), app).serve_forever() else: mulserver = WSGIServer(('0.0.0.0', 8086), app) mulserver.start() def server_forever(): mulserver.start_accepting() mulserver._stop_event.wait() for i in range(cpu_count()): p = Process(target=server_forever) p.start() if __name__ == "__main__": # 单进程 + 协程 run(False) # 多进程 + 协程 #run(True)
python
程序员