使用 PostgreSQL 和 psycopg3 构建 Python 应用程序：开发者指南

作者：Semab Tariq

当我们想到使用 PostgreSQL 和 Python 构建应用程序时，我们最关心的问题是如何使用 Python 与 PostgreSQL 交互。 10 多年来，我们一直依靠一个名为 psycopg2 的可靠适配器。 它为我们提供了良好的服务，在 PostgreSQL 和 Python 之间建立了牢固的连接。 

但是，快速的技术进步促使我们需要 psycopg2 的更高级版本。 与 psycopg2 相比，这种新的选项提供了增强的功能、灵活性以及改进的功能。 这些改进包括服务器端绑定、从一个或多个执行的查询中获取所有结果，以及对二进制通信的支持。 

因此，在本博文中，我们将探讨以下内容

• 在 Python 中构建基本的 CRUD 应用程序

• psycopg2 和 psycopg3 的比较

• psycopg3 中的关键改进

• Timescale 如何简化开发人员的数据库部署 

• 开发者常见问题解答

如何安装 psycopg

二进制安装

如果您只对 psycopg 二进制文件感兴趣，并且想要立即开始开发，我们可以通过以下方式进行： 

pip install "psycopg[binary]"

构建包

如果您更喜欢从源代码构建 psycopg，则需要以下先决条件

• C 编译器

• Python 开发头文件（例如，python3-dev 包）。

• PostgreSQL 客户端开发头文件（例如，libpq-dev 包）。

• pg_config 程序在 PATH 中可用。

在编译 psycopg 包时使用以下命令

pip install "psycopg[c]"

使用 Python 和 psycopg 创建简单的 CRUD 应用程序

在继续使用代码之前，请确保您已在系统上安装了 PostgreSQL，并且它已配置为接受连接。 

按照此链接在您的系统上安装 PostgreSQL： https://postgresql.ac.cn/download/linux/debian/

数据库准备就绪后，创建一个新文件并将以下代码插入其中。

import psycopg
# Connection details (replace with your own)
DATABASE_URL = "postgresql://postgres@localhost:5432/postgres"

def create_item(name, description):
  with psycopg.connect(DATABASE_URL) as conn:
    with conn.cursor() as cur:
      cur.execute("INSERT INTO items (name, description) VALUES (%s, %s)", (name, description))
      conn.commit() # Commit the changes to the database

def read_items():
  with psycopg.connect(DATABASE_URL) as conn:
   with conn.cursor() as cur:
      cur.execute("SELECT * FROM items")
      return cur.fetchall() # Fetch all results as a list of tuples

def update_item(id, name, description):
  with psycopg.connect(DATABASE_URL) as conn:
    with conn.cursor() as cur:
      cur.execute("UPDATE items SET name = %s, description = %s WHERE id = %s", (name, description, id))
      conn.commit()

def delete_item(id):
  with psycopg.connect(DATABASE_URL) as conn:
    with conn.cursor() as cur:
      cur.execute("DELETE FROM items WHERE id = %s", (id,))
      conn.commit()

# Creating/inserting the record
create_item("New Item", "This is a new item.")
items = read_items()
print(items)

# Updating the record
update_item(1, "Updated Item", "This item has been updated.")

# Deleting the record
delete_item(2)

通过 python3 <FILENAME>.py 运行代码。

psycopg2 和 psycopg3 的比较

有关详细的性能比较，请查看我们的 psycopg2 与 psycopg3 性能基准测试。

psycopg 中的增强功能

二进制通信

在 psycopg 中，此功能涉及在 Python 和 PostgreSQL 之间交换二进制数据，例如图像或文件。 当通过标准 %s 占位符传递值时，通常会选择最合适的格式（文本或二进制），但二进制速度最快。 我们可以指定的其他选项是： 

• %b 用于二进制

• %t 用于文本

默认情况下，数据以文本格式返回，但您可以在需要时在 execute 函数中使用 binary=True。

# Sending binary data to the database (using %s):
cur.execute("INSERT INTO images (data) VALUES (%b)", (image_data,))

# Retrieving binary data as is (default):
cur.execute("SELECT data FROM images")
binary_data = cur.fetchone()[0]  # Returns binary data as bytes

# Retrieving data as text:
text_data = cur.fetchone(binary=False)[0]  # Decodes binary data to text

原始查询游标

游标是用于通过执行 SQL 查询与 PostgreSQL 数据库通信的工具。 它们通常由连接的 cursor() 方法创建。

想象一下，您正在通过一个特殊的窗口与图书管理员交谈。 您对着麦克风说话，询问有关书籍的问题（SQL 查询）。 图书管理员在另一边认真倾听，并检索您想要的信息，通过窗口将其呈现出来。 这个窗口充当您的游标。

现在，让我们更深入地了解这个比喻。 就像通过窗口进行对话一样，您可以使用游标进行顺序访问和控制。 您提出的每个问题都会在您继续下一个问题之前得到特定的答案，这反映了控制的、顺序的通信方式。 这类似于进行结构化的对话，其中每个查询都对应一个特定的响应，让您可以一次处理一个查询的结果。

此外，游标是有状态的，这意味着它会记住您在数据集中的位置。 在我们的比喻中，这就像图书管理员记得您已经问过哪些问题以及回答过哪些问题。 这种记忆使每个新问题都能从之前的交互中获益，从而形成连贯而有见地的对话。

在 psycopg 的 3.2 版本中引入的原始查询游标使开发人员能够使用 PostgreSQL 的原生占位符，例如 $1、$2 等，而不是在参数化查询中通常使用的标准 %s 占位符。

from psycopg import connect, RawCursor
with connect(dsn) as conn:
   with RawCursor(conn) as cur:
       cur.execute("SELECT $1, $2", [1, "Hello"])
       assert cur.fetchone() == (1, "Hello")

服务器端绑定

psycopg 中的此功能将查询和参数分别发送到数据库，就像厨师分别收到菜谱和食材一样。 数据库本身会将它们组合在一起并执行查询，就像厨师按照菜谱进行操作一样。

PostgreSQL 提供了一种替代方法来处理 SET，使用 set_config()，以及处理 NOTIFY，使用 pg_notify()。请参阅以下示例。

conn.execute("SELECT set_config('TimeZone', %s, false)", ["UTC"])
conn.execute("SELECT pg_notify(%s, %s)", ["chan", "42"])

from psycopg import sql
cur.execute(
   sql.SQL("INSERT INTO {} VALUES (%s, %s)")
       .format(sql.Identifier('my_table')),
   [10, 20])

psycopg 中的服务器端绑定提供了许多优势，例如性能提升、增强针对 SQL 注入的安全性以及处理二进制数据。

同一查询中的多个语句

在服务器端绑定中，如果您在 SQL 查询中包含参数，则无法在单个 execute() 调用中发送以分号分隔的多个语句。这是因为服务器端绑定需要仔细匹配参数与特定语句，这在一次执行多个语句时会变得很棘手。

如以下示例所示，使用客户端游标来执行此操作：

cur = psycopg.ClientCursor(conn)
>>> cur.execute(
...     "INSERT INTO foo VALUES (%s); INSERT INTO foo VALUES (%s)",
...     (10, 20))

获取所有结果

在 psycopg2 中，如果您运行多个产生多个结果的命令，它只显示最终结果。但是，psycopg 提供了一种更有效的方法来处理多个结果。运行查询后，第一个结果会立即在游标中可用，可以使用标准方法获取。要访问后续结果，可以使用 cursor.nextset() 方法。

cur_pg3.execute("SELECT 1; SELECT 2")
cur_pg3.fetchone()
(1,)

cur_pg3.nextset()
True
cur_pg3.fetchone()
(2,)

cur_pg3.nextset()
None  # no more results

异步操作

psycopg 允许您与数据库通信，而不会中断 Python 程序的主线程。您可以异步地提出问题（查询），这意味着它们会被发送出去，而不会暂停其他所有操作。在等待答案的同时，您的程序可以继续执行其他任务，不会因为等待缓慢的数据库回复而卡住。

这对于提高代码的响应能力和性能非常有用，尤其是在处理长时间运行的查询或频繁与数据库通信时。

AsyncConnection 和 AsyncCursor

psycopg 提供了 Connection 和 Cursor 对象的异步版本，提供非阻塞方法来执行查询、获取结果以及管理事务。

在 psycopg2 中，以下循环演示了 fileno() 和 poll() 方法以及 Python select() 函数在 psycopg 中进行异步操作的使用方式。

def wait(conn):
   while True:
       state = conn.poll()
       if state == psycopg2.extensions.POLL_OK:
           break
       elif state == psycopg2.extensions.POLL_WRITE:
           select.select([], [conn.fileno()], [])
       elif state == psycopg2.extensions.POLL_READ:
           select.select([conn.fileno()], [], [])
       else:
           raise psycopg2.OperationalError("poll() returned %s" % state)

在 psycopg 中，我们可以使用以下示例使用 psycopg 与 PostgreSQL 数据库建立异步连接，异步执行查询，并相应地管理游标。

async with await psycopg.AsyncConnection.connect() as aconn:
   async with aconn.cursor() as cur:
       await cur.execute(...)

使用 COPY TO 和 COPY FROM

在处理 COPY TO 和 COPY FROM 操作时，psycopg 提供了比 psycopg2 更好的功能，它利用了 PostgreSQL 的 COPY 协议，该协议以使用 SQL 技术进行数据加载和修改的效率而闻名。

以下代码将指定表格中的行复制到标准输出，并使其准备好进行进一步处理。

with cur.copy("COPY (SELECT * FROM table_name LIMIT %s) TO STDOUT", (3,)) as copy

使用复制操作，您可以将数据从各种 Python 结构（如元组列表或任何可迭代序列）导入到数据库中。

records = [(10, 20, "hello"), (40, None, "world")]
with cursor.copy("COPY sample (col1, col2, col3) FROM STDIN") as copy:
for record in records:
copy.write_row(record)

如果数据已经以适合复制的方式格式化，则可以使用以下方法将其加载到数据库中：

copy.write(). 
with open("data", "r") as f:
   with cursor.copy("COPY data FROM STDIN") as copy:
       while data := f.read(BLOCK_SIZE):
           copy.write(data)

这就像将文件的块复制粘贴到数据库表格中一样。

连接池

psycopg2 支持连接池，但它不像其继任者 psycopg 提供的功能那样健壮和先进。

安装使用 psycopg 连接池所需的软件包。

pip install psycopg_pool

以下是如何从池中获取连接、执行查询以及在事务提交或回滚后将连接返回到池中以便释放所有池资源的示例。

with ConnectionPool(...) as pool:
   with pool.connection() as conn:
       conn.execute("SELECT something FROM somewhere ...")
       with conn.cursor() as cur:
           cur.execute("SELECT something else...")

   # At the end of the `connection()` context, the transaction is committed
  # or rolled back, and the connection returned to the pool
# At the end of the pool context, all the resources used by the pool are released

如果您的应用程序使用异步代码，您可以使用 AsyncConnectionPool 来实现连接池。

async with AsyncConnectionPool(...) as pool:
   async with pool.connection() as conn:
       await conn.execute("SELECT something FROM somewhere ...")
       with conn.cursor() as cur:
          await cur.execute("SELECT something else...")

管道模式

psycopg2 并不正式包含管道模式，但 psycopg 从 3.1 版本开始引入了它。此模式允许将多个查询发送到 PostgreSQL 服务器，而无需等待每个查询完成才能发送下一个查询，这可以提高处理大量小型频繁查询的应用程序的性能。

在 psycopg 中，管道模式通过 connection.pipeline() 方法支持。

以下是如何使用 conn.pipeline() 在管道模式下执行多个查询的示例。

...     conn.execute("INSERT INTO mytable VALUES (%s)", ["hello"])
...     with conn.cursor() as cur:
...         cur.execute("INSERT INTO othertable VALUES (%s)", ["world"])
...         cur.executemany(
...             "INSERT INTO elsewhere VALUES (%s)",
...             [("one",), ("two",), ("four",)])

管道模式和异步模式有一些相似之处，但在某些方面有所不同。

Timescale：在几分钟内部署您的 PostgreSQL 服务器

为应用程序开发人员设置数据库可能很困难且耗时。正确配置所有内容需要花费很多精力和时间。但是，有了 Timescale，一切都会变得容易得多。开发人员只需点击几下，就能在几分钟内设置好自己的数据库服务器。

这意味着减少了麻烦，开发人员可以将更多时间集中在应用程序开发上。

• 如何将您的数据迁移到 Timescale（3 种方法）

• 额外内容：Psycopg2 与 Psycopg3 性能基准测试

• 使用 Apache Superset 在 PostgreSQL 中进行数据可视化

• 5 种监控 PostgreSQL 数据库的方法

Timescale 提供 30 天的免费试用期，允许您创建最多三个服务，在最初的 30 天使用期间不会产生任何费用。首先，注册 Timescale。

注册并验证电子邮件后，您就可以创建第一个服务了。

目前，有两种可用的服务：

• 时间序列和分析：PostgreSQL，但速度更快。专为基于时间的数据的闪电般快速的摄取和查询而设计。非常适合传感器数据、财务数据、事件或任何其他以时间戳或递增键为索引的数据集。针对性能和成本效益进行了优化，可扩展至大型规模。

• 动态 PostgreSQL: PostgreSQL 拥有无服务器的优势，但没有其问题。非常适合存储元数据、业务信息、客户记录或任何其他传统的关联工作负载。

选择您感兴趣的服务。提交所有详细信息并完成服务创建后，您的数据库服务器将在几分钟内准备就绪。完成后，您将看到连接到数据库的 psql 命令。您还可以下载一个包含有关您的服务的所有详细信息（包括用户名和密码）的纯文本文件。确保将此文件保存在安全的位置。

让我们看看如何更新上面的代码以与托管在 Timescale 上的 tsdb 数据库配合使用。

仅修改 DATABASE_URL，保持代码的其余部分不变。

DATABASE_URL = "postgres://<USERNAME>:<PASSWORD>@<HOST>:<PORT>/tsdb?sslmode=require"

请参阅您之前下载的 <SERVICENAME>-credentials.sql 文件，获取所有必要的信息。

查看 Timescale 文档，详细了解 Timescale 云服务。

开发人员常见问题解答

使用 Python 可以开发哪些类型的应用程序？

Python 可用于开发各种应用程序。

• Web 应用程序

• 数据分析工具

• 人工智能和机器学习应用程序

• 自动化脚本

• 桌面 GUI 应用程序

• 网络服务器等等。

我通过 pip 安装了 psycopg，但我仍然遇到 Python 导入错误。确保您使用的是正确的 Python 环境。确保您已激活虚拟环境或正在使用安装了 psycopg 的正确 Python 环境。此外，确保您安装了 psycopg 包，并使用相同的用户运行您的应用程序，以避免潜在的问题。

我可以在开发中使用 Docker 吗？是的，您可以在开发中使用 Docker。Docker 提供了一个容器化平台，允许您将应用程序及其依赖项打包到容器中。这确保了不同环境之间的一致性，并简化了依赖项管理。Docker 容器可以轻松地在各种系统上部署和运行，为您的应用程序提供一致的开发和部署环境。

如何隐藏数据库凭据？要隐藏 Python 中的数据库凭据，您可以使用环境变量或配置文件。

我可以将 PostgreSQL 用于 Web 和移动应用程序开发吗？是的，PostgreSQL 非常适合 Web 和移动应用程序开发。它提供了可靠性、可扩展性、对复杂查询的支持以及各种数据类型等功能。因此，它可以轻松地用于构建 Web 和移动应用程序。

我可以将 JSON 数据存储在 PostgreSQL 中吗？是的，PostgreSQL 支持存储和查询 JSON 数据。它原生支持 JSON 和 JSONB（二进制 JSON）数据类型，允许您在数据库中存储、检索和操作 JSON 文档。了解更多。

REST API 和 CRUD 之间的区别是什么？CRUD 是一组用于数据操作的操作（创建、读取、更新、删除），而 REST API 是一种用于 Web 服务设计的架构风格，可以包括 CRUD 功能。