如何在Python中读取SQL Server视图？代码无报错却返回空结果

问题背景 #

使用pyodbc驱动从SQL Server视图读取数据时，接口返回空JSON数组（状态码200），但视图后台确有数据；且相同代码查询普通数据库表时运行正常。尝试过pymysql、jdbc等其他库，均未得到预期结果。

原示例代码 #

import logging
import json
import pyodbc
import azure.functions as func


def main(req: func.HttpRequest) -> func.HttpResponse:

    server = '***.database.windows.net'
    database = '**'
    username = '**'
    password = '**'
    driver = '{SQL Server Native Client 11.0}'

cnxn = pyodbc.connect('DRIVER='+driver+';PORT=1433;SERVER='+server +
                      ';PORT=1443;DATABASE='+database+';UID='+username+';PWD=' + password)

if req.method == 'GET':
    result = []
    logging.info('Python HTTP trigger function processed a request.')

    name = req.params.get('name')
    if not terminal_name:
        try:
            req_body = req.get_json()
        except ValueError:
            return func.HttpResponse(
                "Invalid input Json response",
                status_code=401
            )
        else:
            name = req_body.get('name')
            id = req_body.get('id')

    query = f"""
                Select distinct([product_type]), product_grade from  cns_terminal_cockpit.v_terminal_outage 
                where terminal_id ={id} and lower(name) = '{name}' 
                and product_grade <> ''
            """

    cursor = cnxn.cursor()
    try:
        cursor.execute(query)
    except TypeError:
        return func.HttpResponse(
            'Failed. The issue is with the query',
            status_code=402
        )

    for row in cursor.fetchall():
        result.append(
            dict(zip([column[0] for column in cursor.description], row))
        )
    del cnxn

    return func.HttpResponse(
        json.dumps(result, default=str),
        mimetype="application/json",
        status_code=200
    )
else:
    return func.HttpResponse(
        'Wrong request method was posted, please use GET method',
        status_code=403
    )

核心问题及修复方案 #

1. 语法与逻辑错误：未定义变量与缩进问题

• 问题：代码中if not terminal_name:的terminal_name未定义，应为name；同时数据库连接代码缩进错误，位于main函数外部，无法访问函数内的server、database等变量，会导致参数获取逻辑失效，查询条件无意义返回空。
• 修复：将连接代码移入main函数内，修正变量名：

  def main(req: func.HttpRequest) -> func.HttpResponse:
      server = '***.database.windows.net'
      database = '**'
      username = '**'
      password = '**'
      driver = '{SQL Server Native Client 11.0}'
  
      # 连接代码移入函数内部
      cnxn = pyodbc.connect(f'DRIVER={driver};SERVER={server};PORT=1433;DATABASE={database};UID={username};PWD={password}')
  
      if req.method == 'GET':
          result = []
          logging.info('Python HTTP trigger function processed a request.')
  
          name = req.params.get('name')
          # 修正变量名：terminal_name -> name
          if not name:
              try:
                  req_body = req.get_json()
              except ValueError:
                  return func.HttpResponse(
                      "Invalid input Json response",
                      status_code=401
                  )
              else:
                  name = req_body.get('name')
                  id = req_body.get('id')
  

2. SQL连接字符串错误：重复PORT参数

• 问题：连接字符串中重复指定PORT=1433和PORT=1443，导致连接端口混乱，可能引发查询异常。
• 修复：保留正确端口（Azure SQL默认1433），删除重复项：

  cnxn = pyodbc.connect(f'DRIVER={driver};SERVER={server};PORT=1433;DATABASE={database};UID={username};PWD={password}')
  

3. SQL注入风险与参数类型问题：字符串拼接查询

• 问题：用f-string拼接SQL，若name含单引号会导致语法错误，id若为字符串类型未加引号也会出错，直接导致查询条件不匹配返回空。
• 修复：使用pyodbc参数化查询，避免注入同时保证参数类型正确：

  # 用?作为占位符改写查询语句
  query = """
      SELECT DISTINCT([product_type]), product_grade 
      FROM cns_terminal_cockpit.v_terminal_outage 
      WHERE terminal_id = ? AND LOWER(name) = ? 
      AND product_grade <> ''
  """
  # 执行时传入参数元组
  cursor.execute(query, (id, name.lower()))
  

4. 驱动版本兼容性问题

• 问题：SQL Server Native Client 11.0为旧版驱动，对Azure SQL兼容性不足，可能导致视图查询异常。
• 修复：更换为新版ODBC驱动：

  driver = '{ODBC Driver 17 for SQL Server}'
  

5. 视图权限验证

• 问题：确认当前数据库用户对视图cns_terminal_cockpit.v_terminal_outage有SELECT权限，部分场景下普通表权限不自动继承到视图。
• 修复：在SQL Server执行授权语句：

  GRANT SELECT ON cns_terminal_cockpit.v_terminal_outage TO [你的数据库用户名];
  

修复后的完整代码 #

import logging
import json
import pyodbc
import azure.functions as func


def main(req: func.HttpRequest) -> func.HttpResponse:
    server = '***.database.windows.net'
    database = '**'
    username = '**'
    password = '**'
    # 使用新版ODBC驱动
    driver = '{ODBC Driver 17 for SQL Server}'

    try:
        cnxn = pyodbc.connect(f'DRIVER={driver};SERVER={server};PORT=1433;DATABASE={database};UID={username};PWD={password}')
    except pyodbc.Error as e:
        return func.HttpResponse(
            f'Database connection failed: {str(e)}',
            status_code=500
        )

    if req.method == 'GET':
        result = []
        logging.info('Python HTTP trigger function processed a request.')

        name = req.params.get('name')
        id = req.params.get('id')
        # 优先从请求体获取参数
        if not name or not id:
            try:
                req_body = req.get_json()
            except ValueError:
                return func.HttpResponse(
                    "Invalid input JSON response. Please provide 'name' and 'id'",
                    status_code=401
                )
            else:
                name = req_body.get('name')
                id = req_body.get('id')
        
        # 参数校验
        if not name or not id:
            return func.HttpResponse(
                "Missing required parameters: 'name' and 'id' are required",
                status_code=400
            )

        # 参数化查询
        query = """
            SELECT DISTINCT([product_type]), product_grade 
            FROM cns_terminal_cockpit.v_terminal_outage 
            WHERE terminal_id = ? AND LOWER(name) = ? 
            AND product_grade <> ''
        """

        cursor = cnxn.cursor()
        try:
            cursor.execute(query, (id, name.lower()))
        except pyodbc.Error as e:
            return func.HttpResponse(
                f'Query execution failed: {str(e)}',
                status_code=402
            )

        # 获取列名
        columns = [column[0] for column in cursor.description]
        for row in cursor.fetchall():
            result.append(dict(zip(columns, row)))
        
        # 关闭连接
        cursor.close()
        cnxn.close()

        return func.HttpResponse(
            json.dumps(result, default=str),
            mimetype="application/json",
            status_code=200
        )
    else:
        return func.HttpResponse(
            'Wrong request method, please use GET',
            status_code=403
        )

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