博客

Python中为了规范数据库访问的一致性，定义了

DBAPI2.0

，详情见

PEP-0249

。

依据此API规范，JPype的

DBAPI2

模块进行了很好的实现。

JPype的

DBAPI2

是通过访问JDBC来实现DBAPI2.0接口，所以可以通过JDBC连接的数据库，它都能连接。

（JPype 也可以用于直接访问 JDBC，见

Python连接GBase 8s数据库（通过JDBC）

）

下面我们以GBase 8s数据库为例来试用下此模块

安装JPype

pip install jpype1

导入dbapi2模块

from jpype import dbapi2

查看API级别、线程安全级别、参数样式

这三个全局参数是接口规范中要求必须定义的，方便使用者了解实现模块的情况

print('apilevel: ', dbapi2.apilevel)
print('threadsafety: ', dbapi2.threadsafety)
print('paramstyle: ', dbapi2.paramstyle)

输出结果

apilevel:  2.0
threadsafety:  2
paramstyle:  qmark

具体含义见下图PEP-0249的说明

建立数据库连接

import jpype

# 建立连接前需要先启动JVM
jpype.addClassPath('gbasedbtjdbc_3.4.0_2_ca1174.jar')
jpype.startJVM()

# 创建数据库连接
conn = dbapi2.connect(
    dsn='jdbc:gbasedbt-sqli://192.168.1.2:9001/testdb:GBASEDBTSERVER=gbaseserver1;NEWCODESET=UTF8,cp1252,819;DB_LOCALE=zh_CN.utf8;CLIENT_LOCALE=zh_CN.utf8;DBDATE=Y4MD-;',
    driver='com.gbasedbt.jdbc.Driver',
    driver_args={'user': 'gbasedbt', 'password': '111111'}
)

创建游标执行SQL

# 创建游标
cur = conn.cursor()
cur.execute("create table test_t1(a int, b varchar(100))")
cur.execute("insert into test_t1 values(?, ?)", (1, 'gbase8s'))
cur.execute("select * from test_t1")
for row in cur:
    for col in row:
        print(col, type(col))
cur.close()
conn.close()

输出结果

1 <java class 'JInt'>
gbase8s <class 'str'>

注意，dbapi2中采样非自动提交模式，所以上边的代码在运行完成之后，数据库中并没有建表插数据，需要commit之后才可以，如下

# 提交事务
conn.commit()

可以按照如下方式将当前连接修改为自动提交模式

conn._jcx.setAutoCommit(True)

在多线程中使用游标

连接GBase8s数据库时，当游标和连接对象不在同一个线程时，报错

from threading import Thread


def cursor_in_thread(conn_input):
    # 传入connection对象，在线程中创建游标并执行sql
    cur = conn_input.cursor()
    cur.execute("insert into test_t1 values(?, ?)", (1, 'gbase8s'))


t = Thread(target=cursor_in_thread, args=(conn,))
t.start()
t.join()

报错如下：

java.lang.java.lang.ExceptionInInitializerError: java.lang.ExceptionInInitializerError

解决办法：

def cursor_in_thread(conn_input):
    # 传入connection对象，在线程中创建游标并执行sql
    if not jpype.isThreadAttachedToJVM():
        jpype.java.lang.Thread.attach()
        jpype.java.lang.Thread.currentThread().setContextClassLoader(jpype.java.lang.ClassLoader.getSystemClassLoader())
    cur = conn_input.cursor()
    cur.execute("insert into test_t1 values(?, ?)", (1, 'gbase8s'))

数据类型测试

    tables = ['t_byte', 't_text', 't_blob', 't_clob', 't_date', 't_timestamp', ]
    cur.execute('insert into t_byte values(?)', ('中国'.encode('utf8'),))
    cur.execute('insert into t_text values(?)', ('中国'.encode('utf8'),))
    cur.execute('insert into t_blob values(?)', ('中国'.encode('utf8'),))
    cur.execute('insert into t_clob values(?)', ('中国'.encode('utf8'),))
    cur.execute('insert into t_date values(?)', ('2022-06-10',))
    cur.execute('insert into t_timestamp values(?)', ('2022-06-10 12:59:59',))
    for table in tables:
        cur.execute('select * from {}'.format(table))
        row = cur.fetchone()
        col = row[0]
        print('{}，对应python的：'.format(table), type(col))

输出结果

t_byte，对应python的： <class 'bytes'>
t_text，对应python的： <class 'str'>
t_blob，对应python的： <class 'bytes'>
t_clob，对应python的： <class 'str'>
t_date，对应python的： <class 'datetime.date'>
t_timestamp，对应python的： <class 'datetime.datetime'>