DbtPy对GBase 8s数据库的数据类型的支持
DbtPy对数据类型的支持
常用的数据类型如下:
字符类型
内置字符类型包括:char, nchar, varchar, nvarchar 和lvarchar
CHARACTER(n) 和 CHARACTER VARYING(n)这样的别名同样支持
参考代码:test_string_type.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 | create = """ create table tab_string( col1 char(32767), col2 nchar(32767), col3 varchar(255), col4 nvarchar(255), col5 lvarchar(32739) ) """ stmt = DbtPy.exec_immediate(conn, create) insert = """ insert into tab_string(col1, col2, col3, col4, col5) values ('字符字段char类型','本地化字符字段nchar类型','可变长度字符字段varchar类型', '本地化可变长度字符字段nvarchar类型','扩展可变长度字符字段lvarchar类型') """ stmt = DbtPy.exec_immediate(conn, insert) select = "select * from tab_string" stmt = DbtPy.exec_immediate(conn, select) result = DbtPy.fetch_assoc(stmt) while result : print ( "字段1的字节长度为:{} ,值为:\"{}\"" . format ( len (result[ 'col1' ].encode( 'utf8' )),result[ 'col1' ])) print ( "字段2的字节长度为:{} ,值为:\"{}\"" . format ( len (result[ 'col2' ].encode( 'utf8' )),result[ 'col2' ])) print ( "字段3的字节长度为:{} ,值为:\"{}\"" . format ( len (result[ 'col3' ].encode( 'utf8' )),result[ 'col3' ])) print ( "字段4的字节长度为:{} ,值为:\"{}\"" . format ( len (result[ 'col4' ].encode( 'utf8' )),result[ 'col4' ])) print ( "字段5的字节长度为:{} ,值为:\"{}\"" . format ( len (result[ 'col5' ].encode( 'utf8' )),result[ 'col5' ])) result = DbtPy.fetch_assoc(stmt) |
数值类型
内置的数值类型包括:
整型:smallint, integer, int8, bigint
自增长整型:serial, serial8, bigserail
浮点型:smallfloat, float, decimal(P)
精确值型:decimal(P,S)
货币类型:money(P,S)
DEC(p,s)、NUMERIC(p,s)、INT和DOUBLE PRECISION这样的别名同样支持
参考代码:test_numeric_type.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | create = """ create table tab_numeric( col1 serial not null, col2 smallint, col3 int, col4 int8, col5 bigint, col6 smallfloat, col7 float, col8 decimal(32,20), col9 decimal(32), colx money(32,20) ) """ stmt = DbtPy.exec_immediate(conn, create) insert = """ insert into tab_numeric(col1, col2, col3, col4, col5, col6, col7, col8, col9, colx) values (0,32767,2147483647,9223372036854775807,9223372036854775807,1234567890,12345678901234567890,123456789012.123456789,12345678901234567890123456789012,123456789012.123456789) """ stmt = DbtPy.exec_immediate(conn, insert) select = "select * from tab_numeric" stmt = DbtPy.exec_immediate(conn, select) result = DbtPy.fetch_assoc(stmt) while result : print ( "字段 1的类型serail ,值为:{}" . format (result[ 'col1' ])) print ( "字段 2的类型samllint ,值为:{}" . format (result[ 'col2' ])) print ( "字段 3的类型integer ,值为:{}" . format (result[ 'col3' ])) print ( "字段 4的类型int8 ,值为:{}" . format (result[ 'col4' ])) print ( "字段 5的类型bigint ,值为:{}" . format (result[ 'col5' ])) print ( "字段 6的类型smallfloat ,值为:{}" . format (result[ 'col6' ])) print ( "字段 7的类型float ,值为:{}" . format (result[ 'col7' ])) print ( "字段 8的类型decimal(32,20) ,值为:{}" . format (result[ 'col8' ])) print ( "字段 9的类型decimal(32) ,值为:{}" . format (result[ 'col9' ])) print ( "字段10的类型money(32,20) ,值为:{}" . format (result[ 'colx' ])) result = DbtPy.fetch_assoc(stmt) |
日期时间型
日期型:date
日期时间型:datetime [first to last]
间隔类型:interval [first to list]
TIMESTAMP(n)这样的别名在部分版本中支持
interval类型部分支持(实现支持部分间隔类型)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | create = """ create table tab_datetime( col1 serial not null, col2 date, col3 datetime year to day, col4 datetime year to second, col5 datetime year to fraction(5), col6 interval day to minute ) """ stmt = DbtPy.exec_immediate(conn, create) prepare = """ insert into tab_datetime(col1,col2,col3,col4,col5,col6) values(0,?,?,?,?,?) """ stmt = DbtPy.prepare(conn, prepare) DbtPy.bind_param(stmt, 1 , '2023-03-21' ) DbtPy.bind_param(stmt, 2 , '2023-03-21' ) DbtPy.bind_param(stmt, 3 , '2023-03-21 12:34:56' ) DbtPy.bind_param(stmt, 4 , '2023-03-21 12:34:56.98765' ) DbtPy.bind_param(stmt, 5 , '-3 12:13' ) result = DbtPy.execute(stmt) # 不支持interval year to month, day to second等 select = "select * from tab_datetime" stmt = DbtPy.exec_immediate(conn, select) result = DbtPy.fetch_assoc(stmt) while result : print ( "字段 1的类型serail ,值为:{}" . format (result[ 'col1' ])) print ( "字段 2的类型date ,值为:{}" . format (result[ 'col2' ])) print ( "字段 3的类型datetime Y-m-d ,值为:{}" . format (result[ 'col3' ])) print ( "字段 4的类型datetime Y-m-d H:M:S ,值为:{}" . format (result[ 'col4' ])) print ( "字段 5的类型datetime Y-m-d H:M:S.F5 ,值为:{}" . format (result[ 'col5' ])) print ( "字段 6的类型interval d H:M ,值为:{}" . format (result[ 'col6' ])) result = DbtPy.fetch_assoc(stmt) |
大对象和智能大对象类型
大对象类型:byte、text
智能大对象类型:blob、 clob
限制:
text仅支持insert/update,不支持查询
clob仅支持使用函数操作(filetoclob(),locopy(),dbms_lob_new_clob())
byte支持使用bytes操作
blob支持使用bytes和函数操作(filetoblob(),locopy())
dbms_lob_new_clob函数:
1 2 3 | create function if not exists dbms_lob_new_clob (lvarchar) returns clob with ( not variant) external name '$GBASEDBTDIR/extend/excompat.1.0/excompat.bld(dbms_lob_new_clob)' language c; |
clob也可以使用blob相同的操作方式插入,获取时直接获取即可
注:fetch_assoc/fetch_tuple/fetch_both不建议获取2个以上lob类型
参考代码:test_lob_type.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | create = """ create table tab_lob( col1 serial not null, col2 byte, col3 blob, col4 clob, col5 text ) """ stmt = DbtPy.exec_immediate(conn, create) # 使用了dbms_log_new_clob函数将string转换为clob # prepare = "insert into tab_lob(col1,col2,col3,col4,col5) values(0,?,?,dbms_lob_new_clob(?),?)" prepare = "insert into tab_lob(col1,col2,col3,col4,col5) values(0,?,?,?,?)" col2_byte = bytes( 'byte类型输入,最大支持2GB' , encoding = 'UTF-8' ) col3_blob = bytes( 'blob类型输入,最大支持4TB' , encoding = 'UTF-8' ) col4_str = 'clob类型输入' col4_clob = bytes(col4_str, encoding = 'UTF-8' ) col5_text = 'text类型输入' stmt = DbtPy.prepare(conn, prepare) DbtPy.bind_param(stmt, 1 , col2_byte, DbtPy.SQL_PARAM_INPUT) DbtPy.bind_param(stmt, 2 , col3_blob, DbtPy.SQL_PARAM_INPUT) DbtPy.bind_param(stmt, 3 , col4_clob, DbtPy.SQL_PARAM_INPUT) DbtPy.bind_param(stmt, 4 , col5_text, DbtPy.SQL_PARAM_INPUT) result = DbtPy.execute(stmt) # text暂时还不支持获取操作 select = "select col1,col2,col3,col4 from tab_lob" stmt = DbtPy.exec_immediate(conn, select) result = DbtPy.fetch_both(stmt) while result : print ( "字段1的值为:\"{}\"" . format (result[ 0 ])) print ( "字段2的值为:\"{}\"" . format (result[ 1 ].decode( 'UTF-8' ))) print ( "字段3的值为:\"{}\"" . format (result[ 2 ].decode( 'UTF-8' ))) print ( "字段4的值为:\"{}\"" . format (result[ 3 ])) result = DbtPy.fetch_both(stmt) |
布尔类型
布尔类型:boolean
取值范围:'t'/1, 'f'/0, null(None)
参考代码:test_boolean_type.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 | create = """ create table tab_boolean( col1 serial not null, col2 boolean ) """ stmt = DbtPy.exec_immediate(conn, create) prepare = """ insert into tab_boolean(col1,col2) values(0,?) """ stmt = DbtPy.prepare(conn, prepare) DbtPy.bind_param(stmt, 1 , True ) result = DbtPy.execute(stmt) DbtPy.bind_param(stmt, 1 , False ) result = DbtPy.execute(stmt) DbtPy.bind_param(stmt, 1 , None ) result = DbtPy.execute(stmt) select = "select * from tab_boolean" stmt = DbtPy.exec_immediate(conn, select) result = DbtPy.fetch_assoc(stmt) while result : print ( "字段 1的类型serail ,值为:{}" . format (result[ 'col1' ])) print ( "字段 2的类型boolean ,值为:{}" . format (result[ 'col2' ])) result = DbtPy.fetch_assoc(stmt) |
扩展类型LIST
LIST类型:list(TYPE not null)
扩展类型查询结果为bytes类型,需要decode操作
参考代码:test_list_type.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | create = """ create table tab_list( col1 serial not null, col2 LIST(varchar(20) not null) ) """ stmt = DbtPy.exec_immediate(conn, create) prepare = """ insert into tab_list(col1,col2) values(0,?) """ col2_list = None stmt = DbtPy.prepare(conn, prepare) DbtPy.bind_param(stmt, 1 , col2_list, DbtPy.SQL_PARAM_INPUT, DbtPy.SQL_CHAR, DbtPy.SQL_INFX_RC_COLLECTION) col2_list = "LIST{'aaaa','bbbb','cccc'}" result = DbtPy.execute(stmt,(col2_list,)) select = "select * from tab_list" stmt = DbtPy.exec_immediate(conn, select) result = DbtPy.fetch_assoc(stmt) while result : print ( "字段 1的类型serail ,值为:{}" . format (result[ 'col1' ])) # LIST 类型以bytes输出, 需转码成字符串 print ( "字段 2的类型list ,值为:{}" . format (result[ 'col2' ].decode( 'UTF-8' ))) result = DbtPy.fetch_assoc(stmt) |
注意
DbtPy使用的环境变量示例如下:
1 2 3 4 | export GBASEDBTDIR= /opt/gbase8s-odbc-driver export CSDK_HOME= /opt/gbase8s-odbc-driver export PATH=$GBASEDBTDIR /bin :$PATH export LD_LIBRARY_PATH=$GBASEDBTDIR /lib :$GBASEDBTDIR /lib/cli :$GBASEDBTDIR /lib/esql :$LD_LIBRARY_PATH |
参考代码:
test_dbtpy.tar.gz