Python多格式数据读取函数返回值转换为JSON格式实现咨询

现有代码修正说明 #

你当前的sqlread函数存在变量名笔误：参数定义为con_w_sql，但函数内误用了未定义的special_path变量，需要先修正该问题才能正常运行。

场景1：本地直接生成JSON字符串/导出JSON文件 #

改造后支持直接输出JSON格式的代码如下：

import os
import json
import modin.pandas as pd

# 修正变量名错误
def sqlread(con_w_sql, sep='#'):
    con, sql = con_w_sql.split(sep)
    df = pd.read_sql(sql, con)
    return df.head()

readict = {
    ".csv": {"read": pd.read_csv},
    ".tsv": {"read": pd.read_csv},
    ".json": {"read": pd.read_json},
    ".xlsx": {"read": pd.read_excel},
    ".xml": {"read": pd.read_xml},
    ".xls": {"read": pd.read_excel},
    ".hdf": {"read": pd.read_hdf},
    ".sql": {"read": sqlread}
}

def read_any(file):
    _, ext = os.path.splitext(file)
    df = readict[ext]["read"](file)
    # 封装返回结构：包含前100行数据和列类型
    result = {
        # orient参数可按需调整：records为每行一个独立对象，适合前端渲染；table为标准JSON表结构
        "data": df.head(100).to_dict(orient="records"),
        "dtypes": df.dtypes.astype(str).to_dict()
    }
    # 返回JSON字符串，ensure_ascii保证中文正常显示
    return json.dumps(result, ensure_ascii=False)

if __name__ == "__main__":
    file = input("enter the path to the file you want to open : ")
    json_result = read_any(file)
    print(json_result)
    # 如需保存为JSON文件，放开下方注释即可
    # with open("output.json", "w", encoding="utf-8") as f:
    #     f.write(json_result)

场景2：作为HTTP API接口返回JSON响应 #

如果要对外提供API服务，用FastAPI框架实现最简可用的接口示例，先执行pip install fastapi uvicorn安装依赖：

import os
from fastapi import FastAPI
import modin.pandas as pd
from pydantic import BaseModel

app = FastAPI()

# 复用修正后的工具函数和读取配置
def sqlread(con_w_sql, sep='#'):
    con, sql = con_w_sql.split(sep)
    df = pd.read_sql(sql, con)
    return df.head()

readict = {
    ".csv": {"read": pd.read_csv},
    ".tsv": {"read": pd.read_csv},
    ".json": {"read": pd.read_json},
    ".xlsx": {"read": pd.read_excel},
    ".xml": {"read": pd.read_xml},
    ".xls": {"read": pd.read_excel},
    ".hdf": {"read": pd.read_hdf},
    ".sql": {"read": sqlread}
}

# 定义接口请求参数格式
class FileRequest(BaseModel):
    file_path: str

@app.post("/read_file", summary="读取指定文件返回前100行数据和字段类型")
def read_file_api(req: FileRequest):
    _, ext = os.path.splitext(req.file_path)
    df = readict[ext]["read"](req.file_path)
    # 标准API返回结构，FastAPI会自动将字典转为JSON响应
    return {
        "code": 200,
        "msg": "读取成功",
        "data": df.head(100).to_dict(orient="records"),
        "dtypes": df.dtypes.astype(str).to_dict()
    }

# 启动服务后访问 http://127.0.0.1:8000/docs 可直接在可视化页面调试接口
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

补充优化建议 #

• 可新增扩展名不存在、文件读取失败的异常捕获逻辑，提升代码鲁棒性
• 大文件场景可增加返回行数的自定义参数，避免单次响应数据量过大
• 涉及中文内容时注意所有序列化方法都添加ensure_ascii=False参数，防止乱码

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