pyodbc cursor.fetchall()返回嵌套元组结构问题求助

我写了一个处理数据库操作的DBConnector类，其中_run_query方法可执行select、insert等各类查询，但执行select查询时，返回的结果格式不符合预期。我想要的是像["New", "Ongoing", "Closed", "Canceled", "Solved"]这样的普通列表，实际得到的却是[("New",), ("Ongoing",), ("Closed",), ("Canceled",), ("Solved",)]这种嵌套元组的列表。

相关代码如下：

database.connector.py #

import logging
import pyodbc

class DBConnector:
   ...

    def _open_cnxn(self) -> pyodbc.Connection:
        """Opens a connection to the database.

        Returns:
            pyodbc.Connection: Connection object
        """

        self.logger.info(f'Openning connection to "{self.server}".')
        return pyodbc.connect(self.conn_string)

    def _run_query(self, query: str, type: str = "select") -> list:
        """Run the given query

        Args:
            query (str): query string
            type (str, optional): select, insert, update, delete, create. Defaults to "select".

        Returns:
            list: Query result if select | else commits and returns [None]
        """

        try:
            cursor = self._open_cnxn().cursor()
            self.logger.info(f'Executing query: "{query}".')
            cursor.execute(query)
        except Exception as error:
            self.logger.error(
                f'There was an error while trying to execute the query.: "{str(error)}"'
            )

        # DEBUG
        columns = [column[0] for column in cursor.description]
        results = [row for row in cursor.fetchall()]

        print(columns)
        print(results)
        # DEBUG

        match type:
            case "select":
                return results
            case "insert" | "update" | "delete" | "create":
                cursor.commit()
                self.logger.info(f"Commited changes to {columns}")
                return [None]

连接字符串（定义在__init__()中） #

self.conn_string = f"DRIVER={{ODBC Driver 17 for SQL Server}}; \
                             SERVER={self.server}; \
                             DATABASE={self.database}; \
                             UID={self.username}; \
                             PWD={self.password}"

测试模块testing.py #

import configparser
import logging
import sys

from database.connector import DBConnector

secrets = configparser.ConfigParser()
secrets.read("secrets/config.ini")

SERVER = secrets.get("DB_HOMOLOG", "server")
DATABASE = secrets.get("DB_HOMOLOG", "database")
USERNAME = secrets.get("DB_HOMOLOG", "username")
PASSWORD = secrets.get("DB_HOMOLOG", "password")

connector = DBConnector(
    server=SERVER, database=DATABASE, username=USERNAME, password=PASSWORD
)

date_field = "STATUS"
print(connector._run_query("SELECT STATUS FROM TICKETS GROUP BY STATUS"))

这是pyodbc的默认行为：cursor.fetchall()返回的每一行结果都是一个元组，哪怕只查询了单个字段。每个元组对应数据库中的一行，元组内的元素就是该行的各个字段值。你查询的是单个字段STATUS，所以每一行就是仅含一个元素的元组，最终结果自然是元组组成的列表。

方案1：针对单个字段查询转换格式 #

直接修改_run_query方法中获取结果的代码，提取每个元组的第一个元素：

# 替换原results赋值行
results = [row[0] for row in cursor.fetchall()]

这样返回的就是单个值组成的普通列表，完全符合预期。

方案2：兼容单字段与多字段查询 #

如果方法需要同时支持单字段和多字段查询，可以通过判断字段数量动态调整返回格式：

columns = [column[0] for column in cursor.description]
rows = cursor.fetchall()

if len(columns) == 1:
    # 单字段查询，返回普通列表
    results = [row[0] for row in rows]
else:
    # 多字段查询，可选择返回元组列表或更易读的字典列表
    # 字典列表示例：results = [dict(zip(columns, row)) for row in rows]
    results = rows

若选择返回字典列表，结果会变成[{"STATUS": "New"}, {"STATUS": "Ongoing"}, ...]，可读性更强，适合多字段场景。

额外优化建议 #

你的代码中_open_cnxn()每次创建新连接后未关闭，容易造成资源泄漏，建议在try...finally块中管理连接和游标：

def _run_query(self, query: str, type: str = "select") -> list:
    cnxn = None
    cursor = None
    try:
        cnxn = self._open_cnxn()
        cursor = cnxn.cursor()
        self.logger.info(f'Executing query: "{query}".')
        cursor.execute(query)
        
        columns = [column[0] for column in cursor.description]
        rows = cursor.fetchall()
        
        # 结果格式处理逻辑...
        
    except Exception as error:
        self.logger.error(f'There was an error while trying to execute the query.: "{str(error)}"')
        raise  # 重新抛出异常让上层处理
    finally:
        if cursor:
            cursor.close()
        if cnxn:
            cnxn.close()
    
    # 后续返回逻辑...

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