Python 当 pgadmin returns True 时 psycopg2 返回 False

Question

我有一个 python 函数，我想在其中检查 PostgreSQL table 是否存在（对，错）

它不会 return True...即使我登录到同一个数据库并检查 PGAdmin4.. 并得到 True。

我错过了提交吗？我尝试添加 commit() 没有效果。

def __exists_table(self, table_name):
    cursor = self.__get_a_cursor()

    try:
        string_to_execute = "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = '" + table_name + "');"

        cursor.execute(string_to_execute)

        query_results = cursor.fetchall()
        if len(query_results) > 1:
            print("__exists_data got back multiple results, using the first")
        query_results = query_results[0][0]

        return query_results
    except Exception as err:
        print("Exception on __exists_table: " + str(err))
        raise err
    finally:
        cursor.close()

Answer 1

您的代码似乎可以正常工作。

我有一个包含单个 table、table1:

的数据库

$ psql -h localhost
psql (11.6, server 12.1 (Debian 12.1-1.pgdg100+1))
Type "help" for help.

lars=> \d
        List of relations
 Schema |  Name  | Type  | Owner
--------+--------+-------+-------
 public | table1 | table | lars
(1 row)

如果我将您的代码包装在可运行的脚本中，如下所示：

import psycopg2


class DBTest:
    def __init__(self):
        self.db = psycopg2.connect('host=localhost dbname=lars password=secret')

    def __get_a_cursor(self):
        return self.db.cursor()

    def __exists_table(self, table_name):
        cursor = self.__get_a_cursor()

        try:
            string_to_execute = "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = '" + table_name + "');"

            cursor.execute(string_to_execute)

            query_results = cursor.fetchall()
            if len(query_results) > 1:
                print("__exists_data got back multiple results, using the first")
            query_results = query_results[0][0]

            return query_results
        except Exception as err:
            print("Exception on __exists_table: " + str(err))
            raise err
        finally:
            cursor.close()

    def test_exists_table(self, table_name):
        return self.__exists_table(table_name)


db = DBTest()
for table_name in ['table1', 'table2']:
    if db.test_exists_table(table_name):
        print(f'{table_name} exists')
    else:
        print(f'{table_name} does not exist')

运行它产生我期望的输出：

table1 exists
table2 does not exist

话虽如此，我将对您的代码进行以下更改。首先，不要像这样创建查询字符串：

string_to_execute = """SELECT EXISTS(
  SELECT 1 FROM pg_catalog.pg_tables
  WHERE schemaname = 'public' 
    AND tablename = '""" + table_name + "');"

cursor.execute(string_to_execute)

我会让您的数据库驱动程序为您处理参数替换：

string_to_execute = """SELECT EXISTS(
  SELECT 1 FROM pg_catalog.pg_tables
  WHERE schemaname = 'public' 
    AND tablename = %s
  )"""

cursor.execute(string_to_execute, (table_name,))

这样更容易阅读也更安全，因为它会正确引用参数中的任何特殊字符。

Python 当 pgadmin returns True 时 psycopg2 返回 False

Python psycopg2 returning False when pgadmin returns True

python

psycopg2