如何通过双条件动态查询SQLite数据库中指定交易所的股票数据？

你只需要对现有FastAPI代码做3处核心调整即可实现双条件动态查询，全程采用参数化查询避免SQL注入风险：

1. 新增依赖导入 #

首先在代码头部导入需要的类型声明和异常处理类：

from typing import Optional
from fastapi import FastAPI, Request, HTTPException

2. 修改接口参数定义 #

给stock_detail接口新增可选的交易所代码查询参数，支持动态传入：

@app.get("/stock/{ticker}")
def stock_detail(request: Request, ticker: str, exchange_code: Optional[str] = None):

如果要求查询必须传入交易所代码，删掉默认值= None即可。

3. 调整SQL查询逻辑 #

动态拼接WHERE条件，根据是否传入交易所代码决定是否加过滤规则：

# 动态构建查询语句和参数
    base_sql = """
        SELECT security.id, security.country, ticker, full_name, exchange_code FROM security
        LEFT JOIN exchange ON exchange.id = exchange_id WHERE ticker = ?
    """
    params = [ticker]
    if exchange_code:
        base_sql += " AND exchange_code = ?"
        params.append(exchange_code)
    
    cursor.execute(base_sql, tuple(params))
    row = cursor.fetchone()

    # 新增空结果判断，避免后续取id时报错
    if not row:
        raise HTTPException(status_code=404, detail="未查询到对应的股票数据")

完整修改后代码 #

#Import libraries 
import sqlite3, config
from typing import Optional
from sqlite3.dbapi2 import Cursor
from fastapi import FastAPI, Request, HTTPException
from fastapi.templating import Jinja2Templates

#Fast API web framework 

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/")
def index(request: Request):
    connection = sqlite3.connect(config.DB_FILE)
    connection.row_factory = sqlite3.Row
    cursor = connection.cursor()

    #SQLiTE query to call data from database from security table

    cursor.execute("""
        SELECT id, country, ticker, full_name FROM security ORDER BY country
    """)

    rows = cursor.fetchall()

    return templates.TemplateResponse("index.html", {"request": request, "id": id, "stocks" : rows})

@app.get("/stock/{ticker}")
def stock_detail(request: Request, ticker: str, exchange_code: Optional[str] = None):
    connection = sqlite3.connect(config.DB_FILE)
    connection.row_factory = sqlite3.Row
    cursor = connection.cursor()
 
    # 动态构建双条件查询
    base_sql = """
        SELECT security.id, ticker, full_name, exchange_code, country FROM security
        LEFT JOIN exchange ON exchange.id = exchange_id WHERE ticker = ?
    """
    params = [ticker]
    if exchange_code:
        base_sql += " AND exchange_code = ?"
        params.append(exchange_code)
    
    cursor.execute(base_sql, tuple(params))
    row = cursor.fetchone()

    if not row:
        raise HTTPException(status_code=404, detail="未查询到对应的股票数据")

    cursor.execute("""
        SELECT * FROM security_price WHERE security_id = ? ORDER BY date DESC
    """, (row['id'],))

    prices = cursor.fetchall()

    return templates.TemplateResponse("stock_detail.html", {"request": request, "id": id, "stock" : row, "bars" : prices})

使用方式 #

需要指定交易所查询时，直接在请求地址后加查询参数即可，例如：

访问 /stock/AMZN?exchange_code=NASDAQ 就会返回你预期的纳斯达克AMZN的唯一数据
不带exchange_code参数时，保留原有查询逻辑，返回第一个匹配对应代码的股票数据

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