如何在cx_Oracle_async中提取Oracle SQL查询的列名？

方案1：访问包装后的原始同步Cursor #

cx_Oracle_async的AsyncCursorWrapper是对cx_Oracle同步Cursor的异步封装，原始同步Cursor的实例被保存在_cursor属性中，而同步Cursor具备description属性。你可以直接通过这个内部属性获取列名：

修改sql_query方法中的列名提取逻辑：

async def sql_query(self, query: str, return_type=pd.DataFrame):
    """Returns either a DataFrame or a list from a query"""
    try:
        await self.cursor.execute(query)
        result = await self.cursor.fetchall()  # 统一提前获取结果，避免分支变量未定义
        
        if return_type == pd.DataFrame:
            if result:
                # 检查并访问包装后的原始cursor的description属性
                if hasattr(self.cursor, "_cursor") and self.cursor._cursor.description:
                    columns = [desc[0] for desc in self.cursor._cursor.description]
                    return pd.DataFrame(result, columns=columns)
                else:
                    return pd.DataFrame(result)
            else:
                return pd.DataFrame()  # 返回空DataFrame
        elif return_type == list:
            return result
    
    except Exception as e:
        logger.error(f"Error executing SQL query: {e}")
        raise

关键说明 #

• 必须在执行await self.cursor.execute(query)和await self.cursor.fetchall()之后，原始Cursor的description才会被正确填充。
• 用hasattr做属性存在性检查，避免因库版本更新导致内部属性名变更引发新异常。

方案2：通过查询元数据获取列名（备选） #

如果担心依赖内部属性的稳定性，可通过Oracle系统视图查询列名，但这种方式仅适合简单场景（如单表查询），需处理复杂的SQL解析逻辑：

# 示例：仅适配简单单表SELECT查询
async def get_query_columns(self, query: str):
    if not query.strip().upper().startswith("SELECT"):
        return []
    # 简单解析SELECT语句中的列（实际场景需更完善的SQL解析工具）
    import sqlparse
    parsed = sqlparse.parse(query)[0]
    columns = []
    in_select = False
    for token in parsed.tokens:
        if token.ttype == sqlparse.tokens.DML and token.value.upper() == "SELECT":
            in_select = True
            continue
        if token.ttype == sqlparse.tokens.Keyword and token.value.upper() == "FROM":
            break
        if in_select and token.ttype not in (sqlparse.tokens.Punctuation, sqlparse.tokens.Whitespace):
            # 处理列别名情况
            col_name = token.value.split(" AS ")[-1].strip()
            columns.append(col_name)
    return columns

说明 #

这种方式通用性差，仅作为方案1的备选，优先推荐方案1。

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