SqlAlchemy适配SQL Server:如何用DEFAULT替代NULL生成插入语句?
解决SqlModel/SQLAlchemy插入时使用DEFAULT而非NULL的问题
问题场景
使用SqlModel(基于SQLAlchemy与Pydantic的框架)定义了带自动生成字段的模型:
class SequencedBaseModel(BaseModel): sequence_id: str = Field(alias="sequence_id") @declared_attr def sequence_id(cls): return Column( 'sequence_id', VARCHAR(50), server_default=text(f"SELECT '{cls.__tablename__}_'" f" + convert(varchar(10), NEXT VALUE FOR dbo.sequence)")) class Project(SequencedBaseModel, table=True): pass
通过API插入数据时,请求JSON未传入sequence_id:
{ "name": "test_project" }
期望由数据库通过server_default生成该字段值,但SQLAlchemy生成的插入语句将sequence_id设为NULL:
insert into Project (name, sequence_id) values ("test_project", null)
这在SQL Server中触发“无法将NULL插入sequence_id列”的异常,正确的SQL应该使用DEFAULT关键字:
insert into Project (name, sequence_id) values ("test_project", default)
解决方案
方法一:调整模型定义,使用FetchedValue()
修改模型,同时配置Pydantic字段和SQLAlchemy列属性,明确告知框架字段值由服务器生成:
from sqlmodel import Field, SQLModel, declared_attr from sqlalchemy import Column, VARCHAR, text from sqlalchemy.sql import FetchedValue class SequencedBaseModel(SQLModel): # Pydantic层面允许字段为None,但SQLAlchemy层面设为非空 sequence_id: str | None = Field(alias="sequence_id", default=None, nullable=False) @declared_attr def sequence_id(cls): return Column( 'sequence_id', VARCHAR(50), nullable=False, server_default=text(f"SELECT '{cls.__tablename__}_' + convert(varchar(10), NEXT VALUE FOR dbo.sequence)"), server_onupdate=FetchedValue() # 可选,更新时也使用服务器默认逻辑 ) class Project(SequencedBaseModel, table=True): name: str # 补充模型中的name字段
关键说明:
nullable=False确保SQLAlchemy不会将该字段视为可空,避免生成NULL值FetchedValue()告诉SQLAlchemy该字段的值由数据库服务器生成,插入时会使用DEFAULT关键字而非NULL- Pydantic字段设为
str | None并给default=None,是为了兼容请求中不传入该字段的场景
方法二:插入时主动排除该字段
如果不想修改模型结构,可以在插入数据时,从数据对象中移除sequence_id字段,让SQLAlchemy生成的INSERT语句不包含该字段,数据库会自动应用server_default:
from sqlmodel import Session from pydantic import BaseModel # 定义仅包含请求字段的Pydantic模型 class ProjectCreate(BaseModel): name: str # 处理插入逻辑 def create_project(session: Session, project_data: ProjectCreate): # 转换为数据库模型对象 db_project = Project(**project_data.dict()) # 移除sequence_id字段,避免传入NULL if hasattr(db_project, "sequence_id"): delattr(db_project, "sequence_id") session.add(db_project) session.commit() session.refresh(db_project) # 刷新获取数据库生成的sequence_id return db_project
为什么之前的尝试无效
直接设置sequence_id: str = Field(alias="sequence_id", default=sqlalchemy.sql.elements.TextClause('default'))是在Pydantic层面设置默认值,框架会将TextClause('default')当作普通字符串处理,最终插入的是字符串"default"而非SQL关键字DEFAULT,因此无法生效。
内容的提问来源于stack exchange,提问作者Matthias Burger
相关产品推荐
相关产品推荐

