Python - Postgresql varchar 的无效字节序列 - 奇怪的行为

Question

我编写了一个使用 python 线程函数的脚本。我认为问题与线程有关，因为当我从工作线程外部运行查询时，它工作正常。我正在尝试将一些东西插入数据库，但遇到了一些非常时髦的行为。

让我简化一下：

运行这个工程：

cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
data = ("solaris-cdc", resultHOST[0], "UX10", 1,)
sql.execute(cmd, data)

运行这个不起作用：

cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
data = ("solaris-cdc", resultHOST[0], "sdsdsdsdsdsd", 1,)
sql.execute(cmd, data)

以下是字段类型： device = varchar host = varchar ux = varchar units = int

这是我收到的错误：

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/usr/local/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/cstanley/scripts/vip/sun_audit.py", line 37, in workon
    sql.execute(cmd, data)
DataError: invalid byte sequence for encoding "UTF8": 0x86
HINT:  This error can also happen if the byte sequence does not match the encoding expected by the server, which is controlled by "client_encoding".

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/usr/local/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/cstanley/scripts/vip/sun_audit.py", line 37, in workon
    sql.execute(cmd, data)
InternalError: current transaction is aborted, commands ignored until end of transaction block

这里是完整的代码：

#!/usr/local/bin/python2.7
import sys, os, string, threading
import paramiko
import psycopg2
import time

#paramiko.util.log_to_file("sun_audit.log")

getCPU = "/usr/sbin/psrinfo -p"
getMEM = "/usr/sbin/prtconf | grep \"Memory\" | awk '{ print $3 }'"
getHOST = "hostname"

class bcolors:
    MAGENTA = '\033[95m'
    YELLOW = '\033[93m'
    ENDC = '\033[0m'

def workon(host,sql):

    #Connect to each host
    ssh = paramiko.SSHClient()
    key = paramiko.RSAKey.from_private_key_file("/home/cstanley/scripts/vip/cstanley")
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host, username='cstanley', pkey=key)

    #Run Commands
    stdinHOST, stdoutHOST, stderrHOST = ssh.exec_command(getHOST)
    stdinCPU, stdoutCPU, stderrCPU = ssh.exec_command(getCPU)
    stdinMEM, stdoutMEM, stderrMEM = ssh.exec_command(getMEM)

    with threading.Lock():

        resultHOST = stdoutHOST.readlines()
        #print "{0} {0} UX10 1".format(resultHOST[0].rstrip())
        cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
        data = ("solaris-cdc", resultHOST[0], "sdsdsdsdsdsd", 1,)
        sql.execute(cmd, data)

        resultCPU = stdoutCPU.readlines()
        ux40 = (int(resultCPU[0].rstrip()) - 1)
        if ux40 != 0:
            #print "{0} {0} UX40 {1}".format(resultHOST[0].rstrip(),ux40)
            cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
            data = ("solaris-cdc", resultHOST[0], "UX40", ux40,)
            sql.execute(cmd, data)

        resultMEM = stdoutMEM.readlines()
        ux30 = (int(resultMEM[0].rstrip()) / 1024 - 2) / 2
        #print "{0} {0} UX30 {1}".format(resultHOST[0].rstrip(),ux30)
        cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
        data = ("solaris-cdc", resultHOST[0], "UX30", ux30,)
        sql.execute(cmd, data)

        ssh.close()

def main():

    #date = (time.strftime("%Y-%m-%d"))

    #Define our connection string
    conn_string = "host='REMOVED' dbname='REMOVED' user='REMOVED' password='REMOVED' connect_timeout=3"

    # print the connection string we will use to connect
    #print bcolors.MAGENTA + 'Connecting to database\n    ->%s' % (conn_string) + bcolors.ENDC + "\n"

    # get a connection, if a connect cannot be made an exception will be raised here
    conn = psycopg2.connect(conn_string)

    # conn.cursor will return a cursor object, you can use this cursor to perform queries
    sql = conn.cursor()
    print bcolors.YELLOW + "Inserting Solaris information into table.\n" + bcolors.ENDC

    with open('/home/cstanley/scripts/vip/sun_ip') as ip:
        hosts = ip.read().splitlines()

    threads = []
    for h in hosts:
        t = threading.Thread(target=workon, args=(h,sql,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

    conn.commit()
    sql.close()
    conn.close()

if __name__ == "__main__":
    main()

试图弄清楚这里发生了什么。为什么当我输入UX10 时它可以工作，而当我输入sdsdsdsdsdsd 时它不起作用？我什至尝试用solaris-cdc 替换它，就像它在查询的第一部分一样，但这也失败了。这到底是怎么回事！？

Answer 1

错误似乎是在说：当 PostgreSQL 期望给出正确的 UTF-8 编码的 unicode 时，您正试图将一些带有二进制的 Python 2 strs 插入 varchar 列。要么它间歇性地来自resultHOST[0]，要么你的代码中的"sdsdsdsdsdsd" 字符串中有一些不可见的控制字符：

>>> print u"here be ma\x86gic"
here be magic

但真正的原因可能是根据psycopg2 documentation，

游标不是线程安全的：多线程应用程序可以从同一个连接创建多个游标，并且应该使用单个线程中的每个游标。详情请见Thread and process safety。

因此，您应该在每个线程上创建一个新光标。不要将sql 作为参数传递给workon，只需在workon 方法中使用语句创建一个新游标

sql = conn.cursor()

此外，应该不需要锁定，因为使用具有多个游标的单个连接是线程安全的。