FastAPI内调用封装requests的自定义库随机返回空值如何解决？

• 核心问题：新增的/company/{tickers}/dosomethingEvenCooler是异步路由，但内部调用的requests是同步阻塞库。同步代码会直接卡住FastAPI的事件循环，高并发场景下大量请求排队，会出现内部HTTP请求还没拿到远端返回，就触发了超时或者连接被重置，最终拿到空响应。
• 次要问题：requests默认的连接池最大连接数只有10，高并发下连接不够用会直接丢弃请求，没有报错但拿不到返回内容。加打印或者sleep相当于人为降低了并发度，所以失败概率会下降。
• 潜在隐患：第一个接口的SQL拼接逻辑存在语法风险和注入风险，where ticker in { *tickers* }的写法如果入参格式异常，可能会返回空结果。

方案1（最优）：改用异步HTTP客户端适配异步路由 #

替换同步requests库为httpx.AsyncClient，适配FastAPI的异步事件循环，避免阻塞：

# 修改工具库的请求逻辑为异步
import httpx
import pandas as pd

async def requestCompanyReturning(ticker, endpoint):
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(
            f'myurl.com/company/{ticker}/{endpoint}', 
            headers={'Authorization': f'{config.api_key}'}
        )
        r.raise_for_status() # 主动抛出HTTP错误，方便排查异常
        df = pd.read_json(r.content)
        return df

# 对应的dosomething方法同步改异步
async def dosomething(ticker):
    df = await requestCompanyReturning(ticker, "dosomething")
    return df

# 异步路由内直接await调用即可
@app.get("/company/{tickers}/dosomethingEvenCooler")
async def dosomethingcooler(tickers):
    var = await dosomething('ibm')
    # 后续业务逻辑
    return jsonable_encoder(var)

方案2（低改造成本）：把异步路由改为同步路由 #

如果不想修改工具库的同步代码，直接去掉路由的async关键字即可，FastAPI会自动把同步路由放到独立线程池运行，不会阻塞事件循环：

# 去掉async关键字，改成同步路由
@app.get("/company/{tickers}/dosomethingEvenCooler")
def dosomethingcooler(tickers):
    var = dosomething('ibm')
    # 后续业务逻辑
    return jsonable_encoder(var)

方案3（坚持用requests的调整）：调大requests连接池配置 #

如果必须保留requests调用，手动调大连接池上限适配高并发场景：

import requests
from requests.adapters import HTTPAdapter

# 全局初始化复用session，调大连接池
session = requests.Session()
session.mount('http://', HTTPAdapter(pool_connections=100, pool_maxsize=100))
session.mount('https://', HTTPAdapter(pool_connections=100, pool_maxsize=100))

def requestCompanyReturning(ticker, endpoint):
    r = session.get(
        f'myurl.com/company/{ticker}/{endpoint}', 
        headers={'Authorization': f'{config.api_key}'},
        timeout=10
    )
    r.raise_for_status()
    df = pd.read_json(r.content)
    return df

额外优化：修复第一个接口的SQL逻辑 #

用asyncpg原生的参数化查询代替字符串拼接，避免语法错误和SQL注入：

@app.get("/company/{tickers}/dosomething")
async def dosomething(tickers: str):
    # 根据实际入参格式拆分ticker列表，比如逗号分隔则用split(',')
    ticker_list = tickers.split(',')
    query_statement = """
    select *
    from mytable
    where ticker = any($1)
    """
    result_data = await main_db_instance.fetch_rows(query_statement, ticker_list)
    return jsonable_encoder(result_data)

内容的提问来源于stack exchange，提问作者Lostsoul