为提高效率,提问时请提供以下信息,问题描述清晰可优先响应。
【DM版本】: dmSQLAlchemy 2.0.2
【操作系统】:Win10 64Bit
【CPU】: Intel(R) Core(TM) i7-8550U
【问题描述】*:
Python:3.12.9 64Bit
SQLAlchemy:2.0.34
驱动:dmPython 2.5.8
方言包:dmSQLAlchemy 2.0.2
使用pip install dmPython dmSQLAlchemy安装驱动和方言包
使用同步模式连接数据库正常
def new_id() -> str:
return ...
class TestModel(DeclarativeBase):
id: Mapped[Annotated[
str,
mapped_column(String(64), primary_key=True,
default=new_id, doc="主键ID"),
]]
当前使用uuid作为new_id的实现,并在主键字段中使用工厂函数作为default。
执行插入时出现异常。通过源码分析发现方言实现存在以下问题:
文件:sqlalchemy_dm/base.py
行号:1623、1627
函数:DMExecutionContext._setup_ins_pk_from_empty、DMExecutionContext._setup_ins_pk_from_implicit_returning
问题点:调用基类方法时未将值返回
您可以使用最新的版本试试

我刚测试可以正常获取生成的ID,测试代码如下
import uuid from sqlalchemy import Column, String, create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 创建 Base 类 Base = declarative_base() # 定义 Product 表 class Product(Base): __tablename__ = 'Product' # 使用 UUID 作为主键 PRODUCT_ID = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) NAME = Column(String(255), nullable=False) AUTHOR = Column(String(255), nullable=True) PUBLISHER = Column(String(255), nullable=True) PUBLISHTIME = Column(String(255), nullable=True) PRODUCTNO = Column(String(255), nullable=True) SATETYSTOCKLEVEL = Column(String(255), nullable=True) ORIGINALPRICE = Column(String(255), nullable=True) NOWPRICE = Column(String(255), nullable=True) DISCOUNT = Column(String(255), nullable=True) DESCRIPTION = Column(String(1000), nullable=True) PHOTO = Column(String(255), nullable=True) TYPE = Column(String(255), nullable=True) PAPERTOTAL = Column(String(255), nullable=True) WORDTOTAL = Column(String(255), nullable=True) SELLSTARTTIME = Column(String(255), nullable=True) SELLENDTIME = Column(String(255), nullable=True) # 连接达梦数据库 conn_url = 'dm+dmPython://SYSDBA:tZUJfA(sNFp5@localhost:5236' engine = create_engine(conn_url, echo=True) # 创建数据库表(如果不存在) Base.metadata.create_all(engine) # 创建 DBSession DBSession = sessionmaker(bind=engine) # 查询所有数据 def fun_select_all(): session = DBSession() list_product = session.query(Product).all() print('查询所有结果:') for product in list_product: print(f"UUID: {product.PRODUCT_ID}, 名称: {product.NAME}, 作者: {product.AUTHOR}, 出版社: {product.PUBLISHER}") print('') session.close() # 使用 RETURNING 语法插入数据并获取 UUID def fun_insert(): session = DBSession() stmt = Product.__table__.insert().values( PRODUCT_ID=str(uuid.uuid4()), # 生成 UUID NAME='水浒传', AUTHOR='施耐庵,罗贯中', PUBLISHER='中华书局', PUBLISHTIME='2005-4-1', PRODUCTNO='9787101046137', SATETYSTOCKLEVEL='10', ORIGINALPRICE='19', NOWPRICE='14.3', DISCOUNT='7.5', DESCRIPTION='《水浒传》是宋江起义故事在民间长期流传基础上产生出来的,吸收了民间文学的营养。', PHOTO='', TYPE='16', PAPERTOTAL='922', WORDTOTAL='912000', SELLSTARTTIME='2006-03-20', SELLENDTIME='' ).returning(Product.PRODUCT_ID) # 让数据库返回主键 UUID result = session.execute(stmt) new_uuid = result.scalar() # 获取 UUID session.commit() print(f'插入成功, 生成的 UUID: {new_uuid}') session.close() # 更新数据 def fun_update(): session = DBSession() product = session.query(Product).filter(Product.NAME == '水浒传').first() if product: product.NAME = '水浒' session.commit() print(f'更新成功, UUID: {product.PRODUCT_ID}') else: print('未找到该产品') session.close() # 删除数据 def fun_delete(): session = DBSession() product = session.query(Product).filter(Product.NAME == '水浒').first() if product: session.delete(product) session.commit() print(f'删除成功, UUID: {product.PRODUCT_ID}') else: print('未找到该产品') session.close() # 主函数 def main(): fun_select_all() fun_insert() fun_select_all() fun_update() fun_select_all() fun_delete() fun_select_all() if __name__ == '__main__': main()
输出:

编译安装:
cd drivers/python/sqlalchemy
python setup.py install