使用Pandas与SQLAlchemy时列数据类型无法保留的问题求助

方案1：直接用SQL复制表（推荐，高效且完全保留类型） #

无需通过Pandas将数据拉取到Python内存，直接在数据库层面完成表结构与数据的复制，能100%保留所有列的原始数据类型，包括varchar的长度限制，同时性能更优（尤其适合大数据量场景）。

修改你的stg_to_final函数：

def stg_to_final(staging_table, engine, final_table):
    # 先删除已存在的最终表（实现if_exists='replace'的逻辑）
    drop_sql = f"IF OBJECT_ID('{final_table}', 'U') IS NOT NULL DROP TABLE {final_table};"
    # 复制临时表的结构和数据到最终表
    copy_sql = f"SELECT * INTO {final_table} FROM {staging_table};"
    
    with engine.connect() as conn:
        conn.execute(drop_sql)
        conn.execute(copy_sql)
        conn.commit()

方案2：从数据库元数据自动生成dtype字典（适合必须用Pandas中转的场景） #

如果业务逻辑必须通过Pandas读取再写入，可以从INFORMATION_SCHEMA.COLUMNS获取临时表的列类型元数据，自动生成to_sql所需的dtype参数，无需手动逐个指定列类型。

代码实现：

import pandas as pd
import sqlalchemy as sa

def get_staging_dtype(engine, staging_table):
    # 处理带schema的表名（如dbo.stg_table）
    if '.' in staging_table:
        schema_name, table_name = staging_table.split('.', 1)
    else:
        schema_name = 'dbo'
        table_name = staging_table
    
    # 查询临时表的列元数据
    meta_sql = f"""
        SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH
        FROM INFORMATION_SCHEMA.COLUMNS
        WHERE TABLE_SCHEMA = '{schema_name}' AND TABLE_NAME = '{table_name}'
    """
    
    with engine.connect() as conn:
        meta_df = pd.read_sql(meta_sql, conn)
    
    dtype = {}
    for _, row in meta_df.iterrows():
        col_name = row['COLUMN_NAME']
        data_type = row['DATA_TYPE']
        max_len = row['CHARACTER_MAXIMUM_LENGTH']
        
        if data_type == 'bit':
            dtype[col_name] = sa.Boolean()
        elif data_type == 'date':
            dtype[col_name] = sa.Date()
        elif data_type == 'varchar' and max_len is not None:
            dtype[col_name] = sa.VARCHAR(max_len)
        # 可根据实际需要扩展其他数据类型的映射（如int、decimal等）
    
    return dtype

def stg_to_final(staging_table, engine, final_table):
    sql = f"SELECT * FROM {staging_table}"
    df_final = pd.read_sql(sql, engine)
    
    # 自动生成与临时表匹配的dtype字典
    dtype = get_staging_dtype(engine, staging_table)
    
    df_final.to_sql(final_table, con=engine, if_exists='replace', index=False, dtype=dtype)

原方法丢失varchar长度的原因 #

Pandas从SQL Server读取数据时，varchar(n)类型会被转换为Python的str类型；而Pandas默认在写入SQL时，会将str类型映射为varchar(max)（对应SQL Server中CHARACTER_MAXIMUM_LENGTH为-1），因此需要通过指定dtype覆盖这一默认行为。

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