南大通用GBase 8a的Python接口使用样例 – 老紫竹的家 (2025)

本文介绍通过python访问GBase 8a的方法。涉及连接，DDL,DML,DQL，存储过程，加载等常见功能。

版本

python区分2和3， 所以驱动也区分。本文介绍以python2为基础，样例也是py2的格式

gbase-connector-python2-5.0.1.tar.gz

安装

解压缩

[gbase@vm246 python]$ lltotal 44-rw-r--r--. 1 root root 43827 Sep 20 08:51 gbase-connector-python2-5.0.1.tar.gz[gbase@vm246 python]$ tar xvf gbase-connector-python2-5.0.1.tar.gzgbase-connector-python2-5.0.1/gbase-connector-python2-5.0.1/GBaseConnector/gbase-connector-python2-5.0.1/GBaseConnector/errmsg/gbase-connector-python2-5.0.1/GBaseConnector/errmsg/eng/gbase-connector-python2-5.0.1/GBaseConnector/errmsg/eng/client_errmsg.pygbase-connector-python2-5.0.1/GBaseConnector/errmsg/eng/__init__.pygbase-connector-python2-5.0.1/GBaseConnector/errmsg/__init__.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseConnection.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseConstants.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseCursor.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseError.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseErrorCode.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseLogger.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseProtocol.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseSocket.pygbase-connector-python2-5.0.1/GBaseConnector/GBaseUtils.pygbase-connector-python2-5.0.1/GBaseConnector/__init__.pygbase-connector-python2-5.0.1/PKG-INFOgbase-connector-python2-5.0.1/READMEgbase-connector-python2-5.0.1/setup.py[gbase@vm246 python]$

安装(Build+Install)

[root@vm246 gbase-connector-python2-5.0.1]$ python setup.py buildrunning buildrunning build_pycreating buildcreating build/libcreating build/lib/GBaseConnectorcopying GBaseConnector/GBaseConnection.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseConstants.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseCursor.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseError.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseErrorCode.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseLogger.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseProtocol.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseSocket.py -> build/lib/GBaseConnectorcopying GBaseConnector/GBaseUtils.py -> build/lib/GBaseConnectorcopying GBaseConnector/__init__.py -> build/lib/GBaseConnectorcreating build/lib/GBaseConnector/errmsgcopying GBaseConnector/errmsg/__init__.py -> build/lib/GBaseConnector/errmsgcreating build/lib/GBaseConnector/errmsg/engcopying GBaseConnector/errmsg/eng/__init__.py -> build/lib/GBaseConnector/errmsg/engcopying GBaseConnector/errmsg/eng/client_errmsg.py -> build/lib/GBaseConnector/errmsg/eng[root@vm246 gbase-connector-python2-5.0.1]# python setup.py installrunning installrunning buildrunning build_pyrunning install_libcreating /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseConnection.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseConstants.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseCursor.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseError.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseErrorCode.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseLogger.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseProtocol.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseSocket.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/GBaseUtils.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcopying build/lib/GBaseConnector/__init__.py -> /usr/lib/python2.7/site-packages/GBaseConnectorcreating /usr/lib/python2.7/site-packages/GBaseConnector/errmsgcopying build/lib/GBaseConnector/errmsg/__init__.py -> /usr/lib/python2.7/site-packages/GBaseConnector/errmsgcreating /usr/lib/python2.7/site-packages/GBaseConnector/errmsg/engcopying build/lib/GBaseConnector/errmsg/eng/__init__.py -> /usr/lib/python2.7/site-packages/GBaseConnector/errmsg/engcopying build/lib/GBaseConnector/errmsg/eng/client_errmsg.py -> /usr/lib/python2.7/site-packages/GBaseConnector/errmsg/engbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseConnection.py to GBaseConnection.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseConstants.py to GBaseConstants.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseCursor.py to GBaseCursor.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseError.py to GBaseError.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseErrorCode.py to GBaseErrorCode.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseLogger.py to GBaseLogger.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseProtocol.py to GBaseProtocol.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseSocket.py to GBaseSocket.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/GBaseUtils.py to GBaseUtils.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/__init__.py to __init__.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/errmsg/__init__.py to __init__.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/errmsg/eng/__init__.py to __init__.pycbyte-compiling /usr/lib/python2.7/site-packages/GBaseConnector/errmsg/eng/client_errmsg.py to client_errmsg.pycrunning install_egg_infoWriting /usr/lib/python2.7/site-packages/gbase_connector_python-5.0.1-py2.7.egg-info[root@vm246 gbase-connector-python2-5.0.1]#

连接数据库

[root@vm246 python]# cat testCon.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'PASSWORD_OF_ROOT', 'connection_timeout':3600, 'charset':'utf8'} try: conn = connect() conn.connect(**config) print conn except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]#[root@vm246 python]# python testCon.py<GBaseConnector.GBaseConnection.GBaseConnection object at 0x7ff39b92d990>

DDL

[gbase@vm246 python]$ cat testDDL.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() print cur.execute("CREATE table if not exists testPython(id int)") except GBaseError.DatabaseError, err: print err finally: conn.close()[gbase@vm246 python]$gbase> use testdb;Query OK, 0 rows affected (Elapsed: 00:00:00.00)gbase> desc testPython;+-------+---------+------+-----+---------+-------+| Field | Type | Null | Key | Default | Extra |+-------+---------+------+-----+---------+-------+| id | int(11) | YES | | NULL | |+-------+---------+------+-----+---------+-------+1 row in set (Elapsed: 00:00:00.01)gbase>

DML

[gbase@vm246 python]$ cat testDML.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() print cur.execute("CREATE table if not exists testPython(id int)") print cur.execute("insert into testPython values(12345678)") except GBaseError.DatabaseError, err: print err finally: conn.close()[gbase@vm246 python]$ python testDML.pyNoneNone [gbase@vm246 python]$ gccli -pAdmin2O11O531 testdbGBase client 9.5.3.29.23fba856. Copyright (c) 2004-2023, GBase. All Rights Reserved.gbase> select * from testPython;+----------+| id |+----------+| 12345678 |+----------+1 row in set (Elapsed: 00:00:00.01)

DQL 单次

[root@vm246 python]# cat testDQL.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop table if exists testPython") cur.execute("CREATE table if not exists testPython(id int)") cur.execute("insert into testPython values(12345678)") cur.execute("select * from testPython") row = cur.fetchone() print row except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testDQL.py(12345678,)[root@vm246 python]#

DQL 多次

前一个select必须全部读取完毕，否则同一个连接会报错 Unread result found.

[root@vm246 python]# cat testDQL.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop table if exists testPython") cur.execute("CREATE table if not exists testPython(id int)") cur.execute("insert into testPython values(12345678)") cur.execute("select * from testPython") row = cur.fetchall() print row cur.execute("insert into testPython values(22222),(33333)") cur.execute("select * from testPython") row2 = cur.fetchmany(2) print row2 except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testDQL.py[(12345678,)][(12345678,), (22222,)][root@vm246 python]#

一次执行多个SQL

使用multi_stmt=True 参数，否则会报错 Use query_iter for statements with multiple queries.
DDL,DML返回的是None
DQL 返回结果集

DDL/DML

[root@vm246 python]# cat testMultiSQL.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop table if exists testPython;CREATE table if not exists testPython(id int);insert into testPython values(12345678)", multi_stmt=True) cur.execute("select * from testPython") row = cur.fetchall() print row except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testMultiSQL.py[(12345678,), (22222,), (33333,)]

DQL

[root@vm246 python]# cat testMultiDQLSQL.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop table if exists testPython") cur.execute("CREATE table if not exists testPython(id int)") cur.execute("insert into testPython values(12345678)") rss = cur.execute("select * from testPython;select count(*) from testPython", multi_stmt=True) for rs in rss: print rs.fetchall() except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testMultiDQLSQL.py[(12345678,)][(1,)][root@vm246 python]#

批量执行Batch

[root@vm246 python]# cat testPrepared.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop table if exists testPython2") cur.execute("CREATE table if not exists testPython2(id int,name varchar(100))") opfmt = "INSERT INTO testPython2(id, name) VALUES(%s, %s)" rows = [] for i in xrange(0, 100): rows.append((i, "row" + str(i))) cur.executemany(opfmt, rows) cur.execute("select * from testPython2") row = cur.fetchall() print row except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testPrepared.py[(0, u'row0'), (1, u'row1'), (2, u'row2'), (3, u'row3'), (4, u'row4'), (5, u'row5'), (6, u'row6'), (7, u'row7'), (8, u'row8'), (9, u'row9'), (10, u'row10'), (11, u'row11'), (12, u'row12'), (13, u'row13'), (14, u'row14'), (15, u'row15'), (16, u'row16'), (17, u'row17'), (18, u'row18'), (19, u'row19'), (20, u'row20'), (21, u'row21'), (22, u'row22'), (23, u'row23'), (24, u'row24'), (25, u'row25'), (26, u'row26'), (27, u'row27'), (28, u'row28'), (29, u'row29'), (30, u'row30'), (31, u'row31'), (32, u'row32'), (33, u'row33'), (34, u'row34'), (35, u'row35'), (36, u'row36'), (37, u'row37'), (38, u'row38'), (39, u'row39'), (40, u'row40'), (41, u'row41'), (42, u'row42'), (43, u'row43'), (44, u'row44'), (45, u'row45'), (46, u'row46'), (47, u'row47'), (48, u'row48'), (49, u'row49'), (50, u'row50'), (51, u'row51'), (52, u'row52'), (53, u'row53'), (54, u'row54'), (55, u'row55'), (56, u'row56'), (57, u'row57'), (58, u'row58'), (59, u'row59'), (60, u'row60'), (61, u'row61'), (62, u'row62'), (63, u'row63'), (64, u'row64'), (65, u'row65'), (66, u'row66'), (67, u'row67'), (68, u'row68'), (69, u'row69'), (70, u'row70'), (71, u'row71'), (72, u'row72'), (73, u'row73'), (74, u'row74'), (75, u'row75'), (76, u'row76'), (77, u'row77'), (78, u'row78'), (79, u'row79'), (80, u'row80'), (81, u'row81'), (82, u'row82'), (83, u'row83'), (84, u'row84'), (85, u'row85'), (86, u'row86'), (87, u'row87'), (88, u'row88'), (89, u'row89'), (90, u'row90'), (91, u'row91'), (92, u'row92'), (93, u'row93'), (94, u'row94'), (95, u'row95'), (96, u'row96'), (97, u'row97'), (98, u'row98'), (99, u'row99')][root@vm246 python]#

存储过程

[root@vm246 python]# cat testProcedure.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop procedure if exists testPython_proc") cur.execute("create procedure testPython_proc() \ begin \ select 1+2;\ select now();\ end") rss = cur.callproc("testPython_proc") for rs in rss: print rs.fetchall() except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testProcedure.py[(3,)][(datetime.datetime(2023, 9, 22, 9, 30, 31),)][root@vm246 python]#

加载

加载成功行数可以直接用rowcount属性。而忽略行数我测试的版本尚不支持，所以用info做了自行解析来实现的。

[root@vm246 python]# cat testLoad.pyfrom GBaseConnector import connect,GBaseErrorif __name__ == '__main__': config = {'host':'10.10.10.10', 'port':5258, 'database':'testdb', 'user':'root', 'passwd':'Admin2O11O531'} try: conn = connect() conn.connect(**config) cur = conn.cursor() cur.execute("drop table if exists testPython2") cur.execute("CREATE table if not exists testPython2(id int,name varchar(100))") cur.execute("load data infile 'file://10.10.10.10/home/gbase/py.txt' into table testpython2 fields terminated by ','") print cur.info print("Loaded",cur.rowcount) infors=cur.info.split() print("TaskId",infors[1]) print("Skipped",infors[7]) except GBaseError.DatabaseError, err: print err finally: conn.close()[root@vm246 python]# python testLoad.pyTask 3237 finished, Loaded 1 records, Skipped 0 records('Loaded', 1)('TaskId', '3237')('Skipped', '0')[root@vm246 python]#

Post Views: 1,138