LangChain Python - 前端码易

创建时间: 2026-09-01最后更新: 2026-09-01作者: yangbo(2d5d2525a)

1. 认识 Psycopg

Python 程序不能直接操作 PostgreSQL 的数据文件。应用需要数据库驱动来建立连接、发送 SQL、传递参数，并把数据库结果转换成 Python 对象。FastAPI 负责处理 HTTP 请求，Psycopg 则负责它与 PostgreSQL 之间的通信。

安装包名称是 psycopg。对于本地学习环境，可以安装包含预编译实现的 binary 版本：

install-psycopg.bash

1uv add "psycopg[binary]>=3.2"

binary 会安装预编译实现，并自带所需的客户端库，配置最少。生产镜像如果希望使用系统提供的 libpq，可以在准备好编译器、Python 开发头文件和 PostgreSQL 客户端开发文件后选择 psycopg[c]；只安装 psycopg 则使用纯 Python 实现，但系统中仍需有可用的 libpq。它们提供相同的 Python API，区别主要在安装依赖、客户端库来源和性能。

数据库 URL 继续从配置读取：

.env.example

1DATABASE_URL=postgresql://task_app:task_app_dev_password@127.0.0.1:5432/task_app

.env.example 只记录配置模板，实际运行前要在项目根目录创建不提交到仓库的 .env，并填入当前环境的真实连接信息。这段 URL 依次包含协议、用户名、密码、主机、端口和数据库名。如果用户名或密码含有 @、:、/ 等特殊字符，需要先进行 URL 编码。直接使用 Psycopg 时，协议写成 postgresql:// 即可。postgresql+psycopg:// 中的 +psycopg 是 SQLAlchemy 用来选择驱动的方言语法，不能直接传给 Psycopg。

app/core/settings.py

01from pydantic import Field
02from pydantic_settings import BaseSettings, SettingsConfigDict
03
04
05class Settings(BaseSettings):
06    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
07
08    database_url: str = Field(alias="DATABASE_URL")
09
10
11settings = Settings()

后续示例假设 task_app 数据库中已经存在 users、tasks 和 task_status_history 表。若运行时出现 relation does not exist，说明连接已经建立，但目标数据库里还没有对应的表，需要先执行建表 SQL 或数据库迁移。

2. 连接

同步连接的最小写法如下：

sync_connection.py

01import psycopg
02
03from app.core.settings import settings
04
05
06with psycopg.connect(settings.database_url) as connection:
07    with connection.cursor() as cursor:
08        cursor.execute("SELECT current_database(), current_user")
09        row = cursor.fetchone()
10        print(row)

按本文配置，成功后会看到类似 ('task_app', 'task_app') 的结果，两个值分别是当前数据库名和当前用户。若提示 connection refused，先检查 PostgreSQL 是否启动以及主机、端口是否正确；password authentication failed 通常表示用户名或密码不匹配；database does not exist 则表示 URL 中的数据库名还没有创建。

连接表示一次数据库会话，游标负责执行语句和读取结果。离开内层 with 会关闭游标；正常离开连接的 with 时，Psycopg 会提交尚未结束的事务并关闭连接。如果异常离开，则先回滚再关闭连接。因此，连接上下文不仅负责关闭资源，也会影响事务最终是提交还是回滚。

需要特别注意的是，Psycopg 默认没有开启自动提交。即使只执行一条 SELECT，也会自动开始事务，直到 commit()、rollback() 或连接上下文结束。上面的短脚本会很快离开 with，没有问题；长时间存活的连接如果查询后一直不结束事务，就可能在 PostgreSQL 中处于 idle in transaction 状态，继续占用事务快照和相关资源。

只执行互不相关的单条语句时，可以使用 autocommit=True：

autocommit.py

1with psycopg.connect(settings.database_url, autocommit=True) as connection:
2    with connection.cursor() as cursor:
3        cursor.execute("SELECT 1")
4        print(cursor.fetchone())

autocommit=True 表示每条语句各自生效，也适合 CREATE DATABASE、VACUUM 等不能在普通事务块中执行的命令。不过，需要原子性的多步操作仍然要放进 connection.transaction()。例如，把任务改为完成状态和记录状态变更必须一起成功：

complete-task.py

01import psycopg
02from psycopg.rows import dict_row
03
04from app.core.settings import settings
05
06
07def complete_task(
08    connection: psycopg.Connection,
09    task_id: int,
10    operator_id: int,
11) -> dict[str, object] | None:
12    with connection.transaction():
13        with connection.cursor(row_factory=dict_row) as cursor:
14            cursor.execute(
15                """
16                UPDATE tasks
17                SET status = 'done', updated_at = now()
18                WHERE id = %s AND status = 'todo'
19                RETURNING id, title, status
20                """,
21                (task_id,),
22            )
23            task = cursor.fetchone()
24
25            if task is None:
26                return None
27
28            cursor.execute(
29                """
30                INSERT INTO task_status_history (
31                    task_id,
32                    from_status,
33                    to_status,
34                    changed_by
35                )
36                VALUES (%s, 'todo', 'done', %s)
37                """,
38                (task_id, operator_id),
39            )
40
41    return task
42
43
44with psycopg.connect(settings.database_url, autocommit=True) as connection:
45    task = complete_task(connection, task_id=1, operator_id=1)

这里把连接设为自动提交，是为了让它在显式事务之外保持空闲；进入 transaction() 后，两条写入仍属于同一个事务。两条语句都成功才会提交，其中任意一步抛出异常都会一起回滚。若没有找到待完成的任务，函数返回 None，事务中没有变更需要提交。

如果未开启自动提交，连接上的第一条语句通常已经启动外层事务。此时再进入 connection.transaction()，Psycopg 会创建保存点；离开内层上下文只会释放保存点，最终仍要由外层事务提交或回滚。初学时可以先采用上面的组合：长连接使用 autocommit=True，需要保证原子性的业务操作再显式进入 transaction()。

这里的 connection 也不是连接池。每次 connect() 都会建立真实连接，为每个 Web 请求重新握手的成本较高。本篇后面会介绍 Psycopg 自带的连接池用法。

fetchone() 返回下一行或 None，默认的行类型是元组。前面的系统信息查询一定会产生一行，因此可以解包：

tuple-row.py

1database_name, user_name = row

需要字典形式时，可以设置行工厂：

dict-row.py

1from psycopg.rows import dict_row
2
3
4with psycopg.connect(settings.database_url, row_factory=dict_row) as connection:
5    with connection.cursor() as cursor:
6        cursor.execute("SELECT id, title, status FROM tasks ORDER BY id LIMIT 1")
7        task = cursor.fetchone()
8        print(task["title"] if task else "没有任务")

正在验证登录状态

请稍候，验证完成后将继续显示文章内容

首页订阅