如何在Pandas中实现CSV类型转换的一致性方案？

前言 #

我开发了一个读取CSV文件并上传至数据库的工具，流程为：用Pandas将CSV转换为DataFrame，再转为记录后通过SQLAlchemy逐行插入数据库。问题在于Pandas的read_csv函数会自动进行类型推断，导致在为数据库标准化数据时出现不一致的行为。

问题 #

最初的问题是，pandas.read_csv会将数组识别为pd.Object类型，转为记录时会变成字符串，SQLAlchemy处理该字段时会拆分字符串，导致数据库中的数组条目变成{"l","i", "k", "e", " ", "t","h", "i", "s"}这样的形式。此外，原本的空值被到处替换为"NaN"或"nan"。

尝试的解决方案 #

我首先尝试手动捕获并标准化数据，代码会遍历ORM中的所有表列（与CSV列完全一致）并进行手动类型检查。

def parse_body(body, table: sqlalchemy.Table):
    df = pd.read_csv(body)
    df[name] = df[name].apply(lambda x: None if pd.isna(x) else x)
    for name in table.columns.keys():  # Convert types from string to python type
        column_type = table.columns.get(name).type.python_type  # type: ignore
        if column_type in (list, tuple, dict):
            df[name] = df[name].apply(ast.literal_eval)
            continue

        if column_type == date:
            df[name] = df[name].apply(lambda x: datetime.strptime(x, "%Y-%m-%d").date())

            continue
        if column_type == datetime:
            df[name] = df[name].apply(
                lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S.%f")
            )
            continue
    return df.to_dict("records") # off to be inserted

该方案一度有效，但后来遇到了大问题：CSV中有一个需要保存为字符串的数字，read_csv自动将其识别为numpy.float64类型，经过该处理流程后生成的表格如下：

该值用于API数据引用，但这种浮点数格式的字符串导致API调用失败。

重新思考 #

我尝试了以下几种方案来彻底解决问题：

df[name] = df[name].apply(lambda x: None if pd.isna(x) else x)
df[name] = df[name].apply(lambda x: column_type(x))

思路：手动将值转换为对应的Python类型

问题：根据空值检查的位置不同，数据库中会存入"nan"或"None"，且数组需要单独处理，破坏了代码的简洁性。

df[name] = df[name].apply(lambda x: None if pd.isna(x) else x)
df[name] = df[name].astype(column_type)

思路：使用Pandas原生的类型转换功能

问题：存在与上述方案相同的问题，且原生类型转换器不支持List类型（pandas_dtype）。

datatype_map = {k: table.c[k].type.python_type for k in table.columns.keys()}
read_csv(body, dtype=datatype_map)

问题：与上述方案存在相同问题，本质都是类似的解决思路，只是实现主体不同（手动、Pandas原生、Pandas原生）。

最终方案（临时） #

我最终采用了以下方案，目前可以正常运行：

def parse_body(body, table: Table):
    df = pd.read_csv(body, dtype="object")
    for name in table.columns.keys():  # Convert types from string to python type
        column_type = table.columns.get(name).type.python_type  # type: ignore
        logger.info(f"{name} | {df[name].dtype.type} | {column_type}")
        if column_type in (list, tuple, dict):
            df[name] = df[name].apply(lambda x: None if pd.isna(x) else x)
            df[name] = df[name].apply(ast.literal_eval)
            continue

        if column_type == date:
            df[name] = df[name].apply(lambda x: datetime.strptime(x, "%Y-%m-%d").date())
            df[name] = df[name].apply(lambda x: None if pd.isna(x) else x)

            continue
        if column_type == datetime:
            df[name] = df[name].apply(
                lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S.%f")
            )
            df[name] = df[name].apply(lambda x: None if pd.isna(x) else x)
            continue
        df[name] = df[name].astype(column_type)
        df[name] = df[name].apply(lambda x: None if pd.isna(x) or x in ("nan", "NaN") else x)

    logger.info(df)
    return df.to_dict("records")

我在读取CSV时将所有数据类型都转为pandas.object（据我了解这是Pandas最通用的数据类型），之后再手动进行类型转换。我并不在意代码重复或不够优雅，毕竟可以对其进行简化。

真正的问题在于该方案不可持续，代码极其脆弱，任何操作顺序的变更都会导致之前的数据库数据污染问题再次出现。我甚至需要在该函数前添加注释：

# DO NOT TOUCH

最终总会有意外值存入CSV并破坏该函数，届时又要重新寻找更好的解决方案。我认为现在就应该彻底解决这个问题。

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