记录下老版本 python3.5 + sqlalchemy_gbasedbt0.2.4　+　sqlalchemy1.3.8　+　DbtPy3.0.5.6适配

原文链接

# 文件下载

CSDK下载

x86 / arm

GBase8s_3.0.0_1-Linux64-ODBC-Driver

GBase8s_3.0.0_1-Aarch64-ODBC-Driver

sqlalchemy-gbasedbt

sqlalchemy_gbasedbt-0.2.4

DbtPy-3.0.5.6

由于使用交叉编译，通过下载源码进行本地编译

# 编译安装

1、解压GBase8s_3.0.0_1包至/opt/gbase8s-odbc-driver

2、设置环境变量

export CSDK_HOME=/opt/gbase8s-odbc-driver
export LD_LIBRARY_PATH=$CSDK_HOME/lib:$CSDK_HOME/lib/cli:$CSDK_HOME/lib/esql:$LD_LIBRARY_PATH
 

3、设置连接配置

export GBASEDBTDIR=/opt/gbase8s-odbc-driver  
mkdir -p $GBASEDBTDIR/etc
echo "gbase01 onsoctcp 192.168.3.225 9088" > $GBASEDBTDIR/etc/sqlhosts

host port读取的是这个配置

3、解压DbtPy-3.0.5.6.tar.gz，编译

设置编译环境变量，执行python3.5 setup.py build

将编译好的DbtPy.cpython-35m-x86_64-linux-gnu.so，DbtPyDbi.py 放至site-packages下

sqlalchemy_gbasedbt-0.2.4-py3.6.egg使用unzip解压后将sqlalchemy_gbasedbt放至site-packages下

# 使用示例

使用前确保环境变量已导入：
export GBASEDBTDIR=/opt/gbase8s-odbc-driver
export CSDK_HOME=/opt/gbase8s-odbc-driver
export PATH=$GBASEDBTDIR/bin:$PATH
export LD_LIBRARY_PATH=$CSDK_HOME/lib:$CSDK_HOME/lib/cli:$CSDK_HOME/lib/esql:$LD_LIBRARY_PATH

# gbas8s_sqlalchemy_test.py

from sqlalchemy import MetaData, Table, Column, String, create_engine
from sqlalchemy.dialects import registry
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
registry.register("gbasedbt", "sqlalchemy_gbasedbt.dbtdb", "GBasedbtDialect")

import os
# 去掉ODBCINI配置的干扰
os.unsetenv('ODBCINI')

Base = declarative_base()

class User(Base):
   # 表的名字:
   __tablename__ = 'user'
   # 表的结构:
   id = Column(String(20), primary_key=True)
   name = Column(String(20))

# 初始化数据库连接:
# ConStr = 'gbasedbt://:@:/;SERVER='
# 注意：hostname:port 基本无意义，直接 SERVER 匹配 GBASEDBTSQLHOSTS 配置文件中的 GBASEDBTSERVER

USER = 'test_user'
PASSWORD = 'Gbase@123'
DATABASE = 'test'
SERVER = 'gbase01'

ConStr = 'gbasedbt://{user}:{password}@/{db};SERVER={server};DLOC=zh_CN.utf8;CLOC=zh_CN.utf8;DELIMIDENT=y'.format(
   user=USER, password=PASSWORD, db=DATABASE, server=SERVER
)

engine = create_engine(ConStr)
# 删除对象
Base.metadata.drop_all(engine)
# 创建对象
Base.metadata.create_all(engine)

# 创建 DBSession 类型:
DBSession = sessionmaker(bind=engine)

def init_data():
   # 创建 session 对象:
   session = DBSession()
   # 创建新 User 对象:
   new_user = User(id='2', name='测试用户 a20512')
   # 添加到 session:
   session.add(new_user)
   # 提交即保存到数据库:
   session.commit()
   # 关闭 session:
   session.close()

 

def test():
   # 创建 Session:
   session = DBSession()
   user = session.query(User).filter(User.id=='2').one_or_none()
   # 打印类型和对象的 name 属性:
   print('Print user')
   if user :
       print('type:', type(user))
       print('name:', user.name)
   # 执行原生 SQL 语句
   sqlstr = "select tabid,tabname from systables where tabid > :id and owner = :myuser"
   params = {"id" : 1, "myuser": USER}
   result_proxy = session.execute(sqlstr, params)
   result = result_proxy.fetchall()
   print('\nPrint tables of user:')
   for item in result:
       print(item)
   # 关闭 Session:
   session.close()

if __name__ == '__main__':
   init_data()
   test()

# 结果

python3.5 gbas8s_sqlalchemy_test.py
打印如下：

GETTING LAST ROW ID
Print user
type: 
name: 测试用户 a20512

Print tables of user:
(1005, 'user')