热门搜索
分布式
数据库一体机
云数据库
建设银行
保险核心
密集交易
注册

DMExecutionContext_dmPython object has not attribute _is_server_side

00000000000 2022/12/16 1332 5 已解决

环境配置

  • python环境：
    • python 3.9.15
    • fastapi 0.75.0
    • sqlalchemy 1.3.23
    • sqlalchemy-dm 1.4.39
    • dmPython 2.4.5
  • 开发环境操作系统：
    • linux 5.15.0-56-generic x86_64(Ubuntu)
    • dm8_20221020_x86_rh6_64_ent(驱动与客户端工具)
  • 服务端：
    • dm8

问题描述

在以上开发环境下，使用sqlalchemy进行操作达梦数据库，一直报错 DMExecutionContext_dmPython object has not attribute _is_server_side，无论是执行原生sql还是使用orm模型映射，都无法正常使用
本地安装有达梦可视化客户端，连接正常

连接代码

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.engine import create_engine


# dm engine
dm_engine = create_engine("dm+dmPython://%s:%s@%s:%s/?schema=%s", pool_pre_ping=True, echo=False, connect_args={'local_code': 1, 'connection_timeout': 15})
session_dm = sessionmaker(bind=dm_engine)
DMBase = declarative_base(bind=dm_engine)

class get_session:
    def __init__(self, eng: sessionmaker):
        self.eng = eng

    def __call__(self):
        self.eng.configure(autocommit=False, autoflush=False)
        session = self.eng()
        try:
            yield session
        finally:
            session.close()

get_dm_session = get_session(session_dm)

数据表ddl

create table "dm_test" (
    "test_1" varchar2(144) not null,
    "test_2" varchar2(2000),
    "test_3" varchar2(2000),
    "test_4" varchar2(2000),
    "test_5" varchar2(2000),
    "test_6" varchar2(2000),
    "test_7" varchar2(2000),
    "test_8" varchar2(1020) not null,
    constraint "pk_test_1" primary key("test_1")
);

orm模型

from sqlalchemy.sql.schema import Column, PrimaryKeyConstraint, UniqueConstraint
from sqlalchemy_dm.types import VARCHAR2
from sqlalchemy.ext.declarative import declarative_base
from . import dm_engine, DMBase


class test_tab(DMBase):
    __tablename__ = "dm_test"
    __table_args__ = (
        PrimaryKeyConstraint("test_1", name="pk_test_1"),
        UniqueConstraint("test_1")
    )

    test_1 = Column(VARCHAR2(144), nullable=False)
    test_2 = Column(VARCHAR2(2000))
    test_3 = Column(VARCHAR2(2000))
    test_4 = Column(VARCHAR2(2000))
    test_5 = Column(VARCHAR2(2000))
    test_6 = Column(VARCHAR2(2000))
    test_7 = Column(VARCHAR2(2000))
    test_8 = Column(VARCHAR2(2000), nullable=False)


DMBase.metadata.create_all()

api

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from . import test_tab


router = APIRouter(prefix='/test')


@router.get('/dm_test')
async def dm_test(session: Session = Depends(get_dm_session)):
    # result = session.execute("select * from all_tables")
    result = session.query(test_tab).all()
    print(result)
    return None
回答 0
暂无回答
扫一扫
联系客服
logo
logo
logo
鄂Copyright © 达梦国产化应用创新实验室
达梦在线服务平台社区用户管理规范

鄂ICP备18017926号-1