看起来你现在遇到了动态SQL存储过程在并发调用下速度不稳定、偶尔偏慢的问题，结合你的场景（Azure Web Apps、SQL Server、Flask+SQLAlchemy+Flask-Executor），我给你几个不用缓存就能落地的优化方向，都是实际项目里验证过有效的：

1. 重构存储过程：用表值参数（TVP）替代CSV字符串拼接 #

你当前的存储过程直接拼接@names参数，这不仅有SQL注入风险，还会让SQL Server每次都重新编译执行计划（因为每次的CSV字符串内容不同），这是性能波动的核心原因之一。换成表值参数（Table-Valued Parameters）能完美解决这两个问题：

第一步：在SQL Server创建表类型

CREATE TYPE dbo.CompanyNameList AS TABLE (CompanyName NVARCHAR(255) NOT NULL PRIMARY KEY);

第二步：修改存储过程使用TVP

CREATE PROCEDURE spGetDetailsByCompanyNames
    @table_name VARCHAR(255),
    @names dbo.CompanyNameList READONLY
AS
BEGIN
    SET NOCOUNT ON; -- 减少额外的网络传输开销
    DECLARE @SQL AS NVARCHAR(MAX);
    -- 用QUOTENAME避免表名注入风险，用JOIN替代IN提升执行效率
    SET @SQL = N'SELECT * FROM [Master].' + QUOTENAME(@table_name) + N' t
                 INNER JOIN @names n ON t.[Franchisee Name] = n.CompanyName;';
    -- 传递TVP参数给动态SQL，支持执行计划重用
    EXEC sp_executesql @SQL, N'@names dbo.CompanyNameList READONLY', @names = @names;
END;

第三步：Flask端适配TVP传递

在Python函数里，用pyodbc的TableValuedParameter来传递公司列表：

from sqlalchemy import text
import pyodbc

def get_details_by_name(table_name, company_names, dates, db):
    try:
        # 构造表值参数的结构
        tvp = pyodbc.TableValuedParameter(
            'dbo.CompanyNameList',
            [('CompanyName', pyodbc.SQL_NVARCHAR, None, 255)]
        )
        # 将公司列表添加到TVP中
        for name in company_names:
            tvp.append((name,))
        
        query = text("EXEC spGetDetailsByCompanyNames @table_name=:table_name, @names=:names")
        # 绑定参数，适配TVP类型
        query = query.bindparams(
            table_name=table_name,
            names=tvp
        )
        
        with db.connect() as conn:
            return pd.read_sql(query, conn, parse_dates=dates)
    except (sqlalchemy.exc.DatabaseError, TypeError) as e:
        logging.error(f"Query execution failed: {e}")
        return None

2. 优化SQLAlchemy连接池配置 #

并发调用时，连接池不足会导致任务等待获取数据库连接，这也是性能变慢的常见原因。在Flask配置里调整连接池参数：

# 在create_app函数里添加连接池配置
app.config['SQLALCHEMY_POOL_SIZE'] = 5  # 保持的空闲连接数，匹配你的并发任务数
app.config['SQLALCHEMY_MAX_OVERFLOW'] = 10  # 超过pool_size时允许的临时连接数
app.config['SQLALCHEMY_POOL_RECYCLE'] = 3600  # 自动回收超过1小时的空闲连接，避免SQL Server主动关闭

另外，当前的with db.connect()用法是合理的，但确保连接池参数和你的并发任务数匹配（比如你有4个并发任务，pool_size设为5就足够）。

3. 避免SELECT *，只查询需要的列 #

SELECT *会返回所有列，包括你不需要的字段，增加网络传输量和内存消耗。如果前端只需要特定列，直接在存储过程里指定列名：

-- 把SELECT *改成业务需要的列
SET @SQL = N'SELECT [Franchisee Name], [ContractID], [SignDate] FROM [Master].' + QUOTENAME(@table_name) + N' t
             INNER JOIN @names n ON t.[Franchisee Name] = n.CompanyName;';

这能显著减少数据传输的时间，尤其是当表包含大字段（比如TEXT、NVARCHAR(MAX)）时效果更明显。

4. 排查SQL Server端的性能瓶颈 #

（1）添加缺失的索引

如果[Franchisee Name]列没有索引，每次查询都会做全表扫描，速度很慢。在对应的表上创建非聚集索引：

CREATE NONCLUSTERED INDEX IX_Table_FranchiseeName ON [Master].[YourTableName] ([Franchisee Name]);

如果你的查询只需要固定的几列，可以创建覆盖索引，让SQL Server直接从索引取数据，不用回表：

CREATE NONCLUSTERED INDEX IX_Table_FranchiseeName_Covering ON [Master].[YourTableName] ([Franchisee Name])
INCLUDE ([ContractID], [SignDate]); -- 这里放查询需要的其他列

（2）检查Azure SQL Server资源使用率

• 在Azure Portal查看SQL Server的DTU/CPU/内存使用率，如果高峰期超过80%，说明资源不足，需要调整服务层级（比如从Basic升级到Standard）
• 联系DBA查看Query Store，确认是否有其他大查询抢占资源（你提到的同事跑大查询确实会影响，可建议用弹性池做资源隔离）
• 查看存储过程的执行计划，是否存在参数嗅探问题，可在存储过程末尾添加OPTION (RECOMPILE)或者OPTIMIZE FOR UNKNOWN来调整。

5. 调整Flask-Executor的并发数 #

你当前提交了4个并发任务，要确保Executor的工作线程数和数据库连接池大小匹配，避免过多线程导致上下文切换开销：

# 初始化Executor时设置最大线程数，和你的并发任务数一致或略高
executor = Executor(max_workers=4)

或者在create_app里初始化：

executor.init_app(app, max_workers=4)

6. 优化pandas的read_sql调用 #

如果返回的数据量很大，可以用chunksize参数分批读取，减少内存压力：

# 示例：分批读取并合并结果
chunks = []
for chunk in pd.read_sql(query, conn, parse_dates=dates, chunksize=1000):
    chunks.append(chunk)
return pd.concat(chunks, ignore_index=True)

另外，确保parse_dates只指定需要转换的列，不要包含无关列，减少pandas的解析时间。

这些优化里，重构存储过程用TVP是最立竿见影的，不仅解决性能波动，还消除了注入风险。然后结合索引优化和连接池调整，应该能让你的查询速度稳定下来，并且提升整体效率。

备注：内容来源于stack exchange，提问作者Ahad Anjum