python - 使用 pymssql 将字节插入 sql server varbinary(max) 列

Question

我正在尝试使用 Python 3.5 和 pymssql 包将字节或字节数组插入到 Sql Server 表中。我不断收到 pymssql.ProgrammingError尝试插入后。

我已使用以下链接和问题作为引用，但我仍然遇到问题:

Explicit Convert of Varbinary(max) column

Insert binary file into MSSQL db (varbinary) with python pymssql

我已经验证我有权限并且可以将其他数据类型插入到 sql server 数据库中。下面是表结构。

Table Structure

尝试01:

sql_cursor = m_sql_conn.cursor()

byte_test_01 = b'This is a byte test'
hex_01 = '0x'.encode('ascii') + binascii.hexlify(byte_test_01)

string_sql_insert = "INSERT INTO CPBB_DevClusterObjs(str_cluster_id, \
    Covert(varbinary(max), obj_cluster_empty)) \
    VALUES('BatchKm|20|k-means++|1'," + hex_01 + ')'

sql_cursor.execute(string_sql_insert)

结果 01

pymssql.ProgrammingError

hex_01 = b'0x54686973206973206120627974652074657374'

尝试02

sql_cursor = m_sql_conn.cursor()

byte_test_01 = b'This is a byte test'
hex_01 = '0x'.encode('ascii') + binascii.hexlify(byte_test_01)

list_insert_many = [('BatchKm|20|k-means++|1', hex_01)]

string_sql_insert = "INSERT INTO \
    CPBB_DevClusterObjs(str_cluster_id,Covert(varbinary(max), obj_cluster_empty)) \ 
    VALUES (%s,%b)"

sql_cursor.executemany(str_sql_statement, list_insert_many)

结果 02

pymssql.ProgrammingError

hex_01 = b'0x54686973206973206120627974652074657374'

尝试03

我取出 CONVERT() 函数并...

sql_cursor = m_sql_conn.cursor()

byte_test_01 = b'This is a byte test'
hex_01 = '0x'.encode('ascii') + binascii.hexlify(byte_test_01)

string_sql_insert = "INSERT INTO CPBB_DevClusterObjs(str_cluster_id,obj_cluster_empty) \ 
     VALUES ('BatchKm|20|k-means++|1'," \ 
    + hex_01 + ')'

sql_cursor.execute(string_sql_insert)

结果 03

pymssql.ProgrammingError

hex_01 = b'0x54686973206973206120627974652074657374'

我错过了一些东西，但我不知道是什么。我在这里肯定需要一些帮助。

更新01

以下是我在测试方法中的完整代码:

def Test():
    # lists
    list_sql_insert_data_type = ['%s', '%b']
    list_return = list()

    # variables
    string_sql_table = r'CPBB_DevClusterObjs'
    str_sql_error = ''
    user = r'user_me'
    host = r'server_me'
    pswd = r'pswd_me'
    db_name = r'db_me'
    bool_insert_into_table = False

    # sql connection
    list_sql_conn = SqlMethods.SqlGenConnection(user, host, pswd, db_name)
    sql_cursor = list_sql_conn[1].cursor()

    # byte test variables
    byte_test_01 = b'This is a byte test'
    hex_01 = '0x'.encode('ascii') + binascii.hexlify(byte_test_01)

    # create insert string
    string_sql_insert = 'INSERT INTO ' + string_sql_table
    string_sql_insert += '(string_cluster_id, CONVERT(varbinary(max), obj_cluster_empty)) '
    string_sql_insert += "VALUES ('BatchKm|20|k-means++|1'," + "'" + str(hex_01) + "'" + ')'

    # explicit string conversation of hex_01
    string_hex_01 = "'" + str(hex_01) + "'"

    # take out the CONVERT() statement
    #string_sql_insert = 'INSERT INTO ' + string_sql_table
    #string_sql_insert += '(string_cluster_id, obj_cluster_empty) '
    #string_sql_insert += "VALUES ('BatchKm|20|k-means++|1'," + "'" + str(hex_01) + "'" + ')'

    try:
        sql_cursor.execute(string_sql_insert)
    except pymssql.OperationalError:
        str_sql_error = 'Operational error was raised'
    except pymssql.ProgrammingError:
        str_sql_error = 'A program error was raised.'
    except pymssql.Error:
        str_sql_error = 'General error raised.'
    except pymssql.DatabaseError:
        str_sql_error = 'Database error raised.'
    except pymssql.DataError:
        str_sql_error = 'Data error raised.'
    except pymssql.IntegrityError:
        str_sql_error = 'Integrity error raised.'
    except pymssql.InterfaceError:
        str_sql_error = 'Interface error raised.'
    except pymssql.InternalError:
        str_sql_error = 'Internal error raised.'
    except pymssql.NotSupportedError:
        str_sql_error = 'Not supported error raised.'
    except pymssql.StandardError:
        str_sql_error = 'Standard error raised.'
    else:
        bool_insert_into_table = True
        list_sql_conn[1].commit()
    finally:
        list_return.append(bool_insert_into_table)
        list_return.append(str_sql_error)

    # return list
    return list_return

更新 01 结果

我尝试了您最初推荐的内容，但收到错误消息“无法将字节隐式转换为字符串”，结果我将其显式转换为字符串。

而不是 pymssql.ProgrammingError 获取 pymssql.OperationalError。

如果我取出 CONVERT() 函数，我会得到一个 pymssql.ProgrammingError

hex_01 = 0x54686973206973206120627974652074657374

string_hex_01 = "'" + str(hex_01) + "'" = 'b'0x54686973206973206120627974652074657374''

string_sql_insert = INSERT INTO CPBB_DevClusterObjs(string_cluster_id, 
                    CONVERT(varbinary(max), obj_cluster_empty)) VALUES   
                    ('BatchKm|20|k-means++','b'0x54686973206973206120627974652074657374'')

这应该是 string_hex_01 是:

string_hex_01 = str(hex_01)[1:] = '0x54686973206973206120627974652074657374'

这也给出了 pymssql.OperationalError

更新 02

我验证我能够插入到数据库和表中。心理提示:确保我的列名正确(str_cluster_id 和 string_cluster_id)。我仍然收到 pymssql.OperationalError。 CONVERT() 函数如何适应这一点。如果我把它拿出来，我会得到一个 pymssql.ProgrammingError。

已验证的 sql 插入工作:

# test insert into string_cluster_id
string_sql_insert = 'INSERT INTO ' + string_sql_table
string_sql_insert += ' (str_cluster_id) '
string_sql_insert += "VALUES ('BatchKm|20|k-means++|1')"

string_cluster_table = r'CPBB_DevClusterObjs'

更新 03

试图插入以下字符串:

string_sql_insert = 'INSERT INTO CPBB_DevClusterObjs'
string_sql_insert += '(obj_cluster_empty) '
string_sql_insert += "VALUES ('0x5468697320697320612062797‌​4652074657374')"

结果更新03

pymssql.OperationalError

更新 04

成功!!以下是有效的。

string_sql_insert = 'INSERT INTO CPBB_DevClusterObjs'
string_sql_insert += '(obj_cluster_empty) '
string_sql_insert += "VALUES (CONVERT(varbinary(max), '0x5468697320697320612062797‌​4652074657374'))"

更新 04 结果

成功插入 varbinary(max) 列
将发布其他更新，因为我发现还有什么可以工作，什么不可以

_mssql 包和异常处理

为了了解我如何能够意识到我的错误在哪里，我绕了一圈。我使用 _mssql 包重写了我的测试方法，并按照 _http://pymssql.org/en/stable/_mssql_examples.html 的文档中的指示利用了异常处理。

对我来说，关键是要实现如何将异常处理用作对象，例如:

except _mssql.MSSQLDatabaseException as db_e:
    str_sql_error = 'mssql database error raised'
exc_db_number = db_e.number
exc_db_msg = db_e.message

db_e.text 给了我这个信息:

不允许从数据类型 varchar 到 varbinary(max) 的隐式转换。使用 CONVERT 函数运行此查询。DB-Lib 错误消息 20018，严重性 16:一般 SQL Server 错误:检查来自 SQL Server 的消息。

这导致我改变了我需要 CONVERT() 函数来转换数据的地方，而不是试图转换目标数据库中已经是 varbinary(max) 列的列。

有效的插入字符串是:

string_sql_insert = 'INSERT INTO CPBB_DevClusterObjs'
string_sql_insert += '(obj_cluster_empty) '
string_sql_insert += "VALUES (CONVERT(varbinary(max), '0x5468697320697320612062797‌​4652074657374'))"

学过的知识

我是一个“白痴”，并没有确保我的列变量是正确的。确保在发布之前这样做。这将有很大帮助。

将异常处理用作对象。它将帮助您排除错误所在。如果您在缺席十多年后重新开始开发/编码，这将有所帮助。

如果您找不到适合您要完成的任务的答案，请不要害怕寻求帮助。我找到了两个引用，但我无法建立联系。有人会通过这个问题帮助你作为证据。

寻求帮助时尽可能详细和具体。我本来可以在发布我的确切测试代码时做得更好。它可能不会加快这个过程。

测试代码的最终版本

下面是用于将二进制数据插入 Microsoft Sql Server 中的 varbinary(max) 列的测试方法的最终版本。

def Test_01():

    # lists
    list_sql_insert_data_type = ['%s', '%b']
    list_return = list()

    # variables
    string_sql_table = r'CPBB_DevClusterObjs'
    str_sql_error = ''
    user = r'user_me'
    host = r'server_me'
    pswd = r'pswd_me'
    db_name = r'db_me'
    bool_insert_into_table = False

    # sql connection
    list_sql_conn = SqlMethods.SqlGenConnection(user, host, pswd, db_name)
    sql_cursor = list_sql_conn[1].cursor()

    # _mssql connection
    _mssql_conn = _mssql.connect(server = host, user = user, password = pswd, database = db_name)

    # byte test variables
    byte_test_01 = b'This is a byte test'
    hex_01 = '0x'.encode('ascii') + binascii.hexlify(byte_test_01)

    # explicit string conversion of hex_01
    string_hex_01 = "'" + str(hex_01) + "'"
    string_hex_02 = str(hex_01)[1:]

    # create insert string
    #string_sql_insert = 'INSERT INTO CPBB_DevClusterObjs'
    #string_sql_insert += '(str_cluster_id, obj_cluster_empty) '
    #string_sql_insert += "VALUES ('BatchKm|20|k-means++|1','0x5468697320697320612062797‌​4652074657374')"

    # sucess!!  below insert works
    string_sql_insert = 'INSERT INTO CPBB_DevClusterObjs'
    string_sql_insert += '(obj_cluster_empty) '
    string_sql_insert += "VALUES (CONVERT(varbinary(max), '0x5468697320697320612062797‌​4652074657374'))"

    # test insert into string_cluster_id
    #string_sql_insert = 'INSERT INTO ' + string_sql_table
    #string_sql_insert += ' (str_cluster_id) '
    #string_sql_insert += "VALUES ('BatchKm|20|k-means++|1')"

    # take out the CONVERT() statement
    #string_sql_insert = 'INSERT INTO ' + string_sql_table
    #string_sql_insert += '(str_cluster_id, obj_cluster_empty) '
    #string_sql_insert += "VALUES ('BatchKm|20|k-means++|1'," + "'" + str(hex_01) + "'" + ')'

    try:
        sql_cursor.execute(string_sql_insert)
    except pymssql.OperationalError as oe:
        str_sql_error = 'Operational error was raised'
    except pymssql.ProgrammingError:
        str_sql_error = 'A program error was raised.'
    except pymssql.Error:
        str_sql_error = 'General error raised.'
    except pymssql.DatabaseError:
        str_sql_error = 'Database error raised.'
    except pymssql.DataError:
        str_sql_error = 'Data error raised.'
    except pymssql.IntegrityError:
        str_sql_error = 'Integrity error raised.'
    except pymssql.InterfaceError:
        str_sql_error = 'Interface error raised.'
    except pymssql.InternalError:
        str_sql_error = 'Internal error raised.'
    except pymssql.NotSupportedError:
        str_sql_error = 'Not supported error raised.'
    except pymssql.StandardError:
        str_sql_error = 'Standard error raised.'
    else:
        bool_insert_into_table = True
        list_sql_conn[1].commit()
    finally:
        list_return.append(bool_insert_into_table)
        list_return.append(str_sql_error)

    # return list
    return list_return

Answer 1

尝试将您的 string_sql_insert 修改为:

string_sql_insert = "INSERT INTO CPBB_DevClusterObjs(str_cluster_id, \
CONVERT(varbinary(max), obj_cluster_empty)) \
VALUES('BatchKm|20|k-means++|1'," + "'" + hex_01 + "'"')'

你正在插入这个:

INSERT INTO CPBB_DevClusterObjs(str_cluster_id, obj_cluster_empty)     VALUES('BatchKm|20|k-means++|1',0x54686973206973206120627974652074657374)

而不是这个:

INSERT INTO CPBB_DevClusterObjs(str_cluster_id, obj_cluster_empty)     VALUES('BatchKm|20|k-means++|1','0x54686973206973206120627974652074657374')

如果您查看 VALUES，您会明白为什么它会引发错误。

根据 https://github.com/pymssql/pymssql/pull/179/files

def insert_and_select(self, cname, value, vartype, params_as_dict=False):

vartype 是 's'

def test_binary_string(self):
    bindata = '{z\n\x03\x07\x194;\x034lE4ISo'.encode('ascii')
    testval = '0x'.encode('ascii') + binascii.hexlify(bindata)
    colval = self.insert_and_select('data_binary', testval, 's')
    self.typeeq(bindata, colval)
    eq_(bindata, colval)