Alembic迁移更新列时服务器默认值被识别为null的问题

核心问题1：list.append()返回值导致更新为Null #

Python的list.append()是原地修改列表，返回值为None。你在values()里直接用subscription.has_bubble_in_countries.append(Country.NL)，相当于把None赋值给了has_bubble_in_countries，直接触发非空约束错误。

核心问题2：SQLAlchemy未自动填充服务器默认值 #

server_default='{}'是数据库层面的默认值，仅在插入新行且未指定该列值时生效。查询现有数据时，SQLAlchemy不会自动将NULL替换为服务器默认值——如果数据库中存在实际为NULL的行（即使pgAdmin显示为{}，可能存在类型映射差异），读取后会得到None。

修复步骤 #

1. 修正列表更新逻辑 #

不要直接使用append()的返回值，先复制原有列表（或初始化空列表），添加元素后再赋值：

def upgrade():
    connection = op.get_bind()

    for subscription in connection.execute(subscription_old_table.select()):
        if subscription.has_bubble_v1:
            # 处理可能的None值，初始化为空列表
            current_countries = subscription.has_bubble_in_countries or []
            # 复制列表并添加新元素
            new_countries = list(current_countries)
            new_countries.append(Country.NL)
            
            connection.execute(
                subscription_old_table.update().where(
                    subscription_old_table.c.id == subscription.id
                ).values(
                    has_bubble_in_countries=new_countries,
                )
            )
    op.drop_column('Subscription', 'has_bubble_v1')

2. 提前修复数据库中的Null值（可选） #

如果确实存在NULL行，可以先执行一次批量更新，将所有NULL替换为默认空数组：

def upgrade():
    connection = op.get_bind()
    
    # 先修复所有NULL值为默认空数组
    connection.execute(
        subscription_old_table.update()
        .where(subscription_old_table.c.has_bubble_in_countries.is_(None))
        .values(has_bubble_in_countries=[])
    )
    
    # 再执行原有迁移逻辑
    for subscription in connection.execute(subscription_old_table.select()):
        if subscription.has_bubble_v1:
            current_countries = list(subscription.has_bubble_in_countries)
            current_countries.append(Country.NL)
            connection.execute(
                subscription_old_table.update().where(
                    subscription_old_table.c.id == subscription.id
                ).values(has_bubble_in_countries=current_countries)
            )
    op.drop_column('Subscription', 'has_bubble_v1')

3. 优化：使用批量SQL更新（提升效率） #

如果数据量较大，循环逐行更新效率较低，可以用PostgreSQL的ARRAY_APPEND函数直接在数据库层面操作，避免Python层面的列表处理：

from sqlalchemy import func

def upgrade():
    connection = op.get_bind()
    
    # 先修复NULL值
    connection.execute(
        subscription_old_table.update()
        .where(subscription_old_table.c.has_bubble_in_countries.is_(None))
        .values(has_bubble_in_countries=[])
    )
    
    # 批量更新满足条件的行
    connection.execute(
        subscription_old_table.update()
        .where(subscription_old_table.c.has_bubble_v1.is_(True))
        .values(
            has_bubble_in_countries=func.array_append(
                subscription_old_table.c.has_bubble_in_countries,
                Country.NL.value  # 枚举类型需传入对应字符串值
            )
        )
    )
    
    op.drop_column('Subscription', 'has_bubble_v1')

关于Alembic识别服务器默认值的说明 #

Alembic/SQLAlchemy不会自动将查询到的NULL替换为server_default的值，因为server_default是数据库的插入时默认行为，不是查询时的自动转换。如果需要确保查询时获取到默认值，要么在查询时手动处理（如or []），要么提前将数据库中的NULL批量更新为默认值。

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