Python连接Kingbase接口-ksycopg2

新上了一套系统使用人大金仓的Kingbase，需接入现有自动巡检程序，文档记录此次部署ksycopg2过程，以供参考。ksycopg2是Python编程语言的KingbaseES数据库适配器，它的主要特点是Python DB API 2.0 规范的完整实现和线程安全。它拥有客户端和服务端游标，支持异步通信和通知、复制。

一.环境概况

二.环境准备

1.ksycopg2获取

通过地址下载ksycopg2： https://www.kingbase.com.cn/downdriven/index.htm

通过链接下载的驱动支持python 3.10.4，另外libpq库也在驱动包内，随着数据库安装包带的驱动只在python 3.5适配。

2.解压并拷贝ksycopg2

查看Python 3.10模块路径

>>> import sys

>>> sys.path

['', '/usr/local/python3.10/lib/python310.zip', '/usr/local/python3.10/lib/python3.10', '/usr/local/python3.10/lib/python3.10/lib-dynload', '/usr/local/python3.10/lib/python3.10/site-packages']

#将ksycopg2包解压至3.10.4的site-packages内

tar -xvf v8r6_python3.10.4_for_ksycopg2_x86_64.tar.gz -C /usr/local/python3.10/lib/python3.10/site-packages/

3.加载libpq库文件

ksycopg2包内包含了libpq库文件，加载ldconfig内即可

# cat /etc/ld.so.conf

include ld.so.conf.d/*.conf

/usr/local/python3.10/lib/python3.10/site-packages/ksycopg2

执行/usr/sbin/ldconfig 加载

4.验证ksycopg2

Python 3.10.4 (main, Jan 20 2024, 21:29:26) [GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux

Type "help", "copyright", "credits" or "license" for more information.

>>> import ksycopg2

>>> ksycopg2.__version__;

'2.8.5 (dt dec pq3 ext lo64)(for Python3.10 64bit KingbaseV008R006B0001, compiled on Wed Sep 28 15:15:13 2022, commit for *******)'

>>> ksycopg2.__libpq_version__;

120002

以上成功配置了ksycopg2，后面测试执行语句

7.语句验证

以下脚本是几个SQL验证

# -*- coding: utf-8 -*-
import ksycopg2
import datetime

database = "kingbase"
user = "system"
password = "Kingbase"
host = "192.168.1.18"
port = "54321"
failed = 0

def check(name, val):
    if val is None:
        global failed
        failed += 1
    else:
        if isinstance(val, ksycopg2.extensions.connection):
            print("close connection")
            val.close()
        print("test", name, "success !", "\n")


def testConn():
    try:
        conn = ksycopg2.connect(
            "dbname={} user={} password={} host={} port={}".format(database, user, password, host, port))
    # conn.set_session(autocommit=True)
    except Exception as e:
        print(e)
        return None
    else:
        return conn


def testConn2():
    try:
        conn = ksycopg2.connect(database=database, user=user, password=password, host=host, port=port)
        cur = conn.cursor()
        cur.execute("select substr(version(),1,28)")
        rows = cur.fetchall()
        print("database version:", rows[0])

        cur.close()
    except Exception as e:
        print(e)
        return None
    else:
        return conn


def testExecute():
    conn = testConn()
    if conn is not None:
        cur = conn.cursor()

        cur.execute('drop table if exists test_ksy')
        cur.execute('create table test_ksy(id integer, name TEXT)')

        cur.execute("insert into test_ksy values(%s, %s)", (1, "John"))
        cur.execute("insert into test_ksy values(%s, %s)", (2, '中文测试文字'))
        cur.execute("insert into test_ksy values(%s, %s)", (3, '！@#￥%……'))

        cur.close()
        conn.commit()
        conn.close()
        return 0
    else:
        return None


def testSelect():
    conn = testConn()
    if conn is not None:
        cur = conn.cursor()

        cur.execute("select * from test_ksy")
        rows = cur.fetchall()
        for c in cur.description:
            print(c.name, "\t", end="")
        print()

        for row in rows:
            for cell in row:
                print(cell, " ", end="")
            print()

        cur.close()
        conn.commit()
        conn.close()
        return 0
    else:
        return None


def testLob():
    conn = testConn()
    if conn is not None:
        cur = conn.cursor()

        cur.execute('drop table if exists test_lob')
        cur.execute('create table test_lob(id integer, b BLOB, c CLOB, ba bytea)')

        ba = bytearray("中文测试字符bytearray", "UTF8")
        b = bytes('中文测试字符bytes' * 2, "UTF8")
        u = u'中文字unicode' * 3
        s = '中文str' * 4

        cur.execute("insert into test_lob values(%s, %s, %s, %s)", (1, ba, ba, ba))
        cur.execute("insert into test_lob values(%s, %s, %s, %s)", (2, b, b, b))
        cur.execute("insert into test_lob values(%s, %s, %s, %s)", (3, u, u, u))
        cur.execute("insert into test_lob values(%s, %s, %s, %s)", (4, s, s, s))

        cur.execute("select * from test_lob")
        rows = cur.fetchall()

        for row in rows:
            for cell in row:
                if isinstance(cell, memoryview):
                    print(type(cell), cell[:].tobytes().decode('UTF8'), " ", end="")
                else:
                    print(type(cell), cell, " ", end="")
            print()

        cur.close()
        conn.commit()
        conn.close()
        return 0
    else:
        return None


if __name__ == "__main__":
    print("ksycopg2 version:", ksycopg2.__version__ + "libpq version:", ksycopg2.__libpq_version__)
    check("testConn", testConn())
    check("testConn2", testConn2())
    check("testExecute", testExecute())
    check("testSelect", testSelect())
    check("testLob", testLob())
    print("failed case:", failed)

更多数据类型对照，SQL操作请参考Kingbase文档https://help.kingbase.com.cn/v8/index.html