1. PostgreSQL 是什么 #

1.1. 一句话定位 #

PostgreSQL 是一个独立运行的程序，专门负责帮你把数据存好、并且能快速查出来。

PostgreSQL是「独立运行的程序」——它不是一个 Python 库，也不是一个文件格式，而是一个像浏览器、像微信那样常驻在后台的服务。你的程序要跟它打交道，得像访问网站那样「连上去」。 二是「能快速查出来」——存数据本身不难，用记事本也能存，难的是数据涨到几百万条之后还能在几毫秒内找到你要的那一条。数据库大部分的复杂度都花在这件事上。

1.2. 用「记账」类比理解数据库 #

假设你要记录公司所有员工的信息。最土的办法是开一个 Excel，第一行写表头（姓名、邮箱、年龄），下面每一行是一个人。

数据库就是这个 Excel 的加强版，词也几乎能对上：

Excel 里的说法	数据库里的说法	含义
一个工作簿文件	数据库（database）	一整套相关的数据
一张工作表	表（table）	一类东西的集合，比如「所有员工」
一列	列 / 字段（column）	这类东西的某个属性，比如「邮箱」
一行	行 / 记录（row）	具体的一个东西，比如「张三这个人」
筛选、排序、公式	SQL 查询	从数据里问出你想要的答案

那为什么不直接用 Excel？三个 Excel 做不到的事，正好是数据库存在的理由：

1. 多个人同时改不会打架。 Excel 两个人同时改会互相覆盖，数据库能让几十个程序同时读写而不出错。
2. 能拦住错数据。 Excel 里年龄那一列你可以填「abc」，数据库可以规定这一列只能是非负整数，填错直接拒绝。
3. 几百万行也不慢。 Excel 到几十万行就卡了，数据库配上索引后，几百万行里查一条依然是毫秒级。

2. 为什么选 PostgreSQL #

2.1. 它和 SQLite / MySQL 的关系 #

这三个都是关系型数据库，都用 SQL。差别主要在「怎么跑起来」和「适合多大的场景」：

	SQLite	PostgreSQL	MySQL
怎么跑	不用装，就是一个文件	要装、要启动服务	要装、要启动服务
适合	单机小工具、手机 App、做实验	业务系统、复杂查询、要求数据不能错	Web 网站、读多写少
多人同时写	弱（会互相锁）	强	强
学习成本	最低	中等（要理解「服务」的概念）	中等

可以这样判断： 只是本地存点东西、不需要别人访问，用 SQLite 最省事；只要涉及「多个程序同时用同一份数据」，就该上 PostgreSQL 或 MySQL。

2.2. PostgreSQL的作用 #

PostgreSQL 出现在两个地方，都不是可选项而是刚需：

一是让 Agent 的会话能跨进程保留。 教程前面用的 InMemorySaver 把对话历史存在内存里，程序一关就全丢。做真实产品显然不行——用户今天聊到一半，明天回来还得接着聊。换成 PostgreSQL 版的 PostgresSaver 就解决了。

二是让长期记忆能跨会话保留。 用户的姓名、城市、偏好这类信息，应该在所有会话里都记得，这就是 PostgresStore 的活。

3. 前置知识 #

这一节讲的都是后面会反复出现的概念。它们本身不难，但如果没先说清楚，后面每一节都会被卡住。已经有数据库经验的读者可以跳到第 4 节。

3.1. 数据库、表、行、列 #

这四个词的层级关系是从大到小一层套一层：

PostgreSQL 服务（装在你电脑上的那个程序）
   └── 数据库 pgdemo          ← 一个项目通常用一个库
         ├── 表 users         ← 存所有用户
         │     ├── 列: id, name, email, age
         │     ├── 行: (1, '张三', 'zhang@...', 30)
         │     └── 行: (2, '李四', 'li@...', 25)
         └── 表 articles      ← 存所有文章

需要特别注意的是：一个 PostgreSQL 服务里可以有很多个数据库，它们互相隔离。 装好之后会自带一个叫 postgres 的库，通常不在里面放业务数据，而是新建一个自己的库（本文用 pgdemo）。

3.2. SQL：跟数据库说话的语言 #

你不能用 Python 直接命令数据库干活，得用它听得懂的语言，也就是 SQL（读作「S-Q-L」或者「sequel」）。

SQL 的设计初衷是「像说英语一样描述你要什么」，所以读起来相当直白：

-- 从 users 这张表里，选出 name 和 age 两列，只要年龄大于 28 的，按年龄从大到小排
SELECT name, age FROM users WHERE age > 28 ORDER BY age DESC;

只需要先认识五个关键字就可以覆盖了日常九成的操作：

关键字	干什么	记法
SELECT	查	「选出……」
INSERT	增	「插入……」
UPDATE	改	「更新……」
DELETE	删	「删除……」
CREATE	建表、建索引	「创建……」

两个书写约定：关键字习惯大写（SELECT 而不是 select，虽然小写也能跑，但大写更易读）；每条语句用分号 ; 结尾（在 psql 里不写分号它会一直等你输完）。

3.3. 客户端与服务端：为什么要「连接」 #

PostgreSQL 是两个角色在配合：

   你的程序                        PostgreSQL 服务
（客户端 client）                （服务端 server）
        │                                │
        │  ① 连接（带上用户名密码）        │
        │ ─────────────────────────────► │
        │  ② 发一条 SQL                   │
        │ ─────────────────────────────► │
        │                                │ ③ 真正去磁盘上找数据
        │  ④ 把结果送回来                  │
        │ ◄───────────────────────────── │

服务端就是你装的那个 PostgreSQL，它一直在后台跑，默认在 5432 这个端口上等着别人来连。客户端是任何想用数据的程序：psql 命令行是一个客户端，你的 Python 脚本也是一个客户端。

理解这一点，很多报错就自然解释得通了：「连接被拒绝」意味着服务端没在跑或者端口不对，跟你的 SQL 写得对不对完全没关系。

3.4. 连接串（DSN） #

客户端连服务端时要说清楚「连哪台机器上的哪个库、用什么身份」。这一串信息通常写成一个字符串，叫连接串或 DSN（Data Source Name）。本文全程用这一个：

postgresql://postgres:postgres@localhost:5432/pgdemo
└────┬───┘   └───┬──┘ └───┬──┘ └───┬───┘ └┬─┘ └──┬─┘
     │           │        │        │      │      └── 数据库名
     │           │        │        │      └───────── 端口，PostgreSQL 默认 5432
     │           │        │        └──────────────── 主机，localhost 表示本机
     │           │        └───────────────────────── 密码
     │           └────────────────────────────────── 用户名
     └────────────────────────────────────────────── 协议，固定这么写

这六段里最容易搞错的是前两段。 postgres:postgres 看起来像重复，其实前一个是用户名、后一个是密码——安装时默认创建的管理员用户就叫 postgres，而密码是你装的时候自己设的。如果你装的时候设了别的密码，把后面那个 postgres 换成你的密码即可。

3.5. 事务：为什么「提交」不能省 #

事务（transaction）是「要么全做完，要么一件都不做」的一组操作。

经典例子是转账：从 A 扣 100 块，给 B 加 100 块。这两步必须捆在一起——如果扣完钱程序崩了，加钱那步没执行，这 100 块就凭空消失了。事务保证的就是这种情况不会发生：崩了就整体回退，账面依然是对的。

事务最直接的影响是你写完数据必须「提交」，否则等于没写：

① 开始事务（psycopg 会自动帮你开）
      │
② INSERT 一行            ← 此时只有你自己看得见
      │
      ├── conn.commit()   ← 提交：改动正式生效，别人也能看到了
      │
      └── conn.rollback() ← 回滚：改动全部撤销，像没发生过

「忘了 commit 导致数据没存进去」是第一名的坑，第 7.3 节会用可运行的例子演示这个现象。

3.6. 索引：为什么查询会慢 #

假设一本 500 页的书，你要找「PostgreSQL」这个词在哪一页。

没有索引，就只能从第 1 页翻到第 500 页，一页页看——数据库里这叫全表扫描（Seq Scan）。有索引，就是翻到书末尾的索引页，直接看到「PostgreSQL ... 见第 37、112 页」，跳过去就行。

数据库的索引就是这个道理：额外维护一份「值 → 在哪一行」的对照表，用空间换时间。

这里有两个容易误解的地方，第 9 节会用实测数据说明：

• 索引不是免费的。 它占磁盘空间，而且每次增删改数据都要顺带更新索引，所以写入会变慢一点。不该建的地方建了索引是负担。
• 有索引不代表一定会用。 如果一个查询要返回表里大部分的行，数据库会主动放弃索引直接全表扫——因为那样反而更快。

3.7. 主键、唯一、非空、检查：四种约束 #

约束（constraint）就是你交给数据库执行的规矩，违反了它就拒绝写入。

而约束是数据库层面的，无论谁从哪个程序、甚至手动用 psql 写进来，都绕不过去。这是数据不出错的最后一道防线。

四种最常用的：

约束	写法	作用	典型用途
主键	PRIMARY KEY	唯一且不能为空，一张表只能有一个	每行的身份证号，通常是 id
唯一	UNIQUE	不允许重复值	邮箱、手机号
非空	NOT NULL	必须填	姓名这类必填项
检查	CHECK (条件)	值必须满足条件	年龄不能是负数

第 6.5 节的建表脚本会把这四种全用上，第 7.6 节会演示违反它们时会抛什么异常。

3.8. psql 和 SQL 不是一回事 #

这个区别很小但极易混淆，单独说一下。

• SQL 是标准语言，SELECT * FROM users; 这种，任何客户端都能发，数据库来执行。
• psql 的反斜杠命令是 psql 这个命令行工具自带的快捷方式，比如 \dt 列出所有表。它们不是 SQL，只有在 psql 里能用，在 Python 代码里写 \dt 会直接报错。

判断方法很简单：以反斜杠 \ 开头的都是 psql 专属命令，不是 SQL。

4. 安装与第一次连接 #

这一节的目标很具体：让你的电脑上真的跑起一个 PostgreSQL，并且用两种方式各连上一次——命令行工具 psql 连一次，Python 脚本连一次。

之所以要连两次，是因为它们排查问题的用途不同。以后遇到「程序连不上」，你可以先用 psql 试一下：psql 能连上说明数据库没问题，是代码或连接串的事；psql 也连不上，那就是数据库本身或网络的事。这个二分法能省掉大量瞎猜的时间。

装服务端有两条路，选一条即可，后面的连接串完全一样（localhost:5432、用户 postgres、密码 postgres）：

装法	适合谁	本机会长什么
Windows 安装包（4.1 节）	想用本机 psql、当普通软件装	一个开机自启的 Windows 服务
Docker（4.2 节）	已经有 Docker Desktop，不想装系统服务	一个名叫 pg-tutorial 的容器

不要两条同时开。 它们都会占用 5432 端口，后开的那个会直接起不来。

下面以 Windows 为主（本文的实测环境就是 Windows）。macOS 和 Linux 上 Docker 命令相同；安装包的点法不同，但连上之后的操作完全一样。

4.1. Windows 安装要点 #

到 PostgreSQL 官方下载页 下载 EnterpriseDB 提供的安装包，一路下一步即可。安装过程中有三个地方要留意，其他都可以用默认值：

步骤	怎么选	为什么
组件选择	至少勾上 PostgreSQL Server 和 Command Line Tools	前者是数据库本体，后者提供 psql 命令
设置密码	记牢这个密码	这是 postgres 用户的密码，之后每次连接都要用；本文假设你设的是 postgres
端口	默认 5432 别改	所有工具默认都找这个端口，改了以后处处要额外指定

装完之后 PostgreSQL 会作为 Windows 服务自动启动，也就是开机就在后台跑，不需要你每次手动打开什么窗口。

走这条路的话，直接看 4.3 节确认 psql 能用。已经决定用 Docker 的，跳过安装包，读下一小节。

4.2. 用 Docker 安装 #

Docker 解决的是「我本机不想装一个开机自启的数据库服务」。它把已经装好的 PostgreSQL 打成镜像，你启动一个容器，数据库就在里面跑。对本文后面所有示例来说，它和安装包没有区别：Python 仍然连 localhost:5432，用户名密码仍然是 postgres / postgres。

如果你还没装过 Docker，只为这一个软件去装 Docker Desktop，未必比 4.1 节更省事——那就走安装包。这一节假定你已经能在 PowerShell 里跑 docker 命令（Windows 上通常来自 Docker Desktop，并且要先把 Desktop 打开，让引擎真正跑起来）。

先确认引擎活着：

# 只问版本，不启动任何容器
docker version

能看到 Server 那一段版本号就可以。如果报「无法连接 Docker 引擎」或根本找不到 docker 命令，先打开 Docker Desktop，等托盘图标不再转圈，再试一次。

4.2.1. 第一次启动 #

下面这条命令做四件事：拉官方镜像、设好密码、把容器的 5432 映射到你电脑的 5432、把数据放到 Docker 管理的卷里（删容器也不会丢数据）。

在 PowerShell 里执行（反引号 ` 是 PowerShell 的换行符，整段当作一条命令）：

# --name 给容器起个固定名字，后面 start / stop / exec 都靠它
# -e POSTGRES_PASSWORD 是官方镜像的必填项，没有密码容器会拒绝启动
# -e POSTGRES_USER 可省略，默认就是 postgres；写上是为了和本文连接串对齐
# -p 5432:5432 左边是你电脑的端口，右边是容器内 PostgreSQL 的端口
# -v 名字:路径 是命名卷：数据存在 Docker 里，不绑到某个 Windows 文件夹
#    PostgreSQL 18 镜像必须挂到 /var/lib/postgresql，不要写成旧版的 .../data
# -d 后台运行；postgres:18 和本文安装包同大版本，方便对照
docker run --name pg-tutorial -e POSTGRES_PASSWORD=postgres -e POSTGRES_USER=postgres -p 5432:5432 -v pg-tutorial-data:/var/lib/postgresql -d postgres:18

第一次会花一点时间下载镜像。成功后它只打印一长串容器 ID，没有别的输出，这是正常的。

用 docker ps 看它是否在跑：

# --format 只打印我们关心的几列，避免默认表格太宽
docker ps --filter "name=pg-tutorial" --format "table {{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}"

正常时类似：

NAMES          IMAGE         STATUS         PORTS
pg-tutorial    postgres:18   Up 12 seconds  0.0.0.0:5432->5432/tcp

0.0.0.0:5432->5432/tcp 的意思是：你电脑上的 5432 已经转到容器里的 PostgreSQL。所以后面的连接串仍然是 localhost:5432，不必写成某个容器 IP。

4.2.2. 在容器里用 psql 连一次 #

Docker 这条路本机通常没有 psql 命令，这不是装失败。客户端就在容器里，用 docker exec 进到容器里去跑：

# exec 表示「在已经运行的容器里再执行一条命令」
# -it 让你能交互输入（要进 psql 提示符时需要）
# 后面的 psql -U postgres -d postgres 和 4.5 节完全相同
docker exec -it pg-tutorial psql -U postgres -d postgres

提示符同样变成 postgres=#。先问一下版本，再 \q 退出：

postgres=# SELECT version();
                                    version
-------------------------------------------------------------------------------
 PostgreSQL 18.x on x86_64-pc-linux-gnu, compiled by gcc ...
(1 row)

postgres=# \q

只想执行一条就退出、不进交互界面时，加 -c：

# 容器里的 Linux 环境默认就是 UTF-8，这里一般不会碰到 4.4 节那种 Windows 乱码
docker exec pg-tutorial psql -U postgres -d postgres -c "SELECT version();"

密码已经写在启动容器的环境变量里，docker exec 走的是容器内部的本地连接，不会再问你密码。

4.2.3. 日常开关：第二次起不要再 docker run #

docker run 只在第一次用。容器已经存在时再 run 一次，会报名字冲突：

Error response from daemon: Conflict. The container name "/pg-tutorial" is already in use.

以后每天这样开、关：

# 启动已经创建过的容器（不重新拉镜像、不重建数据）
docker start pg-tutorial

# 用完先停，数据还在命名卷里
docker stop pg-tutorial

想看启动日志（比如密码没设导致立刻退出）：

# --tail 20 只看最后 20 行；卡住时把 --tail 去掉往上翻
docker logs --tail 20 pg-tutorial

练习数据可以全部扔掉时：

# 先停再删容器。命名卷还在：下次用同一条 docker run（同样的 -v 名）会接着用旧数据
docker stop pg-tutorial
docker rm pg-tutorial
# 练习数据也要清零时，再删卷（卷名就是 run 时 -v 冒号左边那个）
docker volume rm pg-tutorial-data

4.2.4. 用 Compose 把命令固定下来（可选） #

docker run 参数一多就容易抄漏。可以在项目目录放一个 docker-compose.yml，以后只记两条短命令。文件内容：

# Compose 会按这个文件创建同名项目下的服务
services:
  # 服务名随便起；下面 container_name 才是 docker ps 里看到的名字
  postgres:
    # 与 4.2.1 节同一张官方镜像
    image: postgres:18
    # 固定容器名，方便 docker exec
    container_name: pg-tutorial
    environment:
      # 必填；和本文连接串里的密码一致
      POSTGRES_PASSWORD: postgres
      # 和本文连接串里的用户名一致
      POSTGRES_USER: postgres
    ports:
      # 本机 5432 -> 容器 5432
      - "5432:5432"
    volumes:
      # 18 必须挂到 /var/lib/postgresql，不要写成 /var/lib/postgresql/data
      - pg-tutorial-data:/var/lib/postgresql

# 命名卷声明；不写这一段 Compose 也会自动建，写上更清楚
volumes:
  pg-tutorial-data:

在这个文件所在的目录执行：

# 后台启动（没有容器就创建，有就启动）
docker compose up -d

# 停止容器，默认不删卷，数据还在
docker compose down

down 后面加上 -v 才会把 pg-tutorial-data 一起删掉，练习库要清零时再用。

docker run 和 Compose 不要混着用同一个容器名。已经用 4.2.1 节 run 出了 pg-tutorial 的，就继续 docker start / stop；要用 Compose，先 docker stop pg-tutorial 再 docker rm pg-tutorial（卷可以留着），然后再 docker compose up -d。

4.2.5. 三个最常见的失败 #

端口被占用。 本机已经用安装包跑着 PostgreSQL，或上一次容器还在，再 run 会看到类似：

Bind for 0.0.0.0:5432 failed: port is already allocated

处理：要么停掉 Windows 的 PostgreSQL 服务（服务名一般带 postgresql），要么 docker stop pg-tutorial。不要改成 5432 来「两个都开」——本文后面所有连接串都写死了 5432。

**挂载路径写错。 18 的官方镜像改成了 /var/lib/postgresql，请照抄上面的路径。

Docker Desktop 没启动。 docker run 报无法连接引擎时，先打开 Desktop，等它就绪，不要反复换镜像版本。

装好之后，4.3 节里没有本机 psql 可以跳过；4.5 节用 docker exec 代替 psql；4.6 节的 Python 脚本一行都不用改。

4.3. 确认装好了 #

安装完先做这一步，能省掉后面一堆莫名其妙的排查。

走 Docker、本机没有 psql 的： 不必装客户端，用 4.2.2 节的 docker exec 验证即可，跳过下面这条。

走安装包的： 打开命令行，看版本号：

# --version 只是打印版本号，不连接数据库，所以不需要密码
# 能打印出版本说明命令行工具装好了，而且加进了 PATH
psql --version

本机的输出：

psql (PostgreSQL) 18.4

如果提示「找不到 psql 命令」，走安装包的说明安装时没勾 Command Line Tools，或者 PostgreSQL 的 bin 目录没进 PATH。最快的解决办法是把它手动加上，默认路径是 C:\Program Files\PostgreSQL\18\bin。走 Docker 的则用 4.2.2 节的 docker exec，不必强行装本机客户端。

4.4. Windows 中文乱码：先做这一步 #

Windows 命令行默认不是 UTF-8 编码，而 PostgreSQL 存的是 UTF-8，两边对不上就会看到一堆乱码。在本机用 psql 之前先切一下编码：

# chcp 是 Windows 的「切换代码页」命令，65001 就是 UTF-8
# 加 >nul 是把它自己那句提示信息藏掉，让输出干净些
chcp 65001 >nul

这个设置只对当前这个命令行窗口有效，关掉窗口就没了，所以每开一个新窗口都要重来一次。

走 Docker、只用 docker exec 在容器里跑 psql 的，容器内部已经是 UTF-8，这一步可以跳过。在 Windows 窗口里直接敲 psql 时才必须切。

没做这一步会看到什么？这是同一条查询，没切编码时的输出：

 count
     3
(1 �м�¼)

最后那行「(1 记录)」被显示成了乱码。切了编码之后就正常了。第 12.2 节还会讲一种切编码也救不了的乱码，以及怎么应对。

4.5. 用 psql 连上去 #

psql 是官方自带的命令行客户端，学 PostgreSQL 绕不开它。连接命令：

# -U 指定用哪个用户连，postgres 是安装时默认创建的管理员
# -d 指定连哪个数据库，postgres 是自带的那个库
# 回车后会提示 Password for user postgres:，输入你安装时设的密码
psql -U postgres -d postgres

走 Docker、本机没有 psql 时，把上面那条换成 4.2.2 节的写法，进去之后提示符和反斜杠命令完全一样：

docker exec -it pg-tutorial psql -U postgres -d postgres

连上之后提示符会变成 postgres=#，表示「已经在数据库里了，可以发 SQL 了」。走 Docker 的 docker exec 走的是容器内部连接，不会再问密码。

刚开始最容易在这里卡住的两点（本机 psql）：

一是输密码时屏幕上什么都不显示。这是命令行的安全设计，不是卡住了，照常输完按回车即可。

二是想退出不知道怎么退。输入 \q 然后回车（注意是反斜杠加 q）。按 Ctrl+C 只能中断当前正在输的那行，退不出去。

如果你嫌每次都输密码麻烦，可以先设一个环境变量：

# PGPASSWORD 是 PostgreSQL 官方认的环境变量，设了之后 psql 不再提示输密码
# 同样只对当前窗口有效
set PGPASSWORD=postgres

这个办法方便，但密码会明文留在命令行历史里。

4.6. 用 Python 连上去 #

uv add "psycopg[binary]"

装好驱动之后，最小的可运行程序是这样，先确认能连通：

# psycopg 是 PostgreSQL 的 Python 驱动，装的是 3.x 版本
import psycopg

# 连接串：协议://用户名:密码@主机:端口/数据库名
# 这里连的是自带的 postgres 库，因为我们还没建自己的库
DSN = "postgresql://postgres:postgres@localhost:5432/postgres"

# connect() 建立连接；用 with 包起来，块结束时会自动关闭连接
with psycopg.connect(DSN) as conn:
    # 游标（cursor）是「执行 SQL、取结果」的把手，也用 with 自动清理
    with conn.cursor() as cur:
        # version() 是 PostgreSQL 的内置函数，返回服务端的版本信息
        cur.execute("SELECT version()")
        # fetchone() 取回一行；结果是元组，所以用 [0] 取第一列
        print(cur.fetchone()[0])

本机安装包的输出：

PostgreSQL 18.4 on x86_64-windows, compiled by msvc-19.44.35227, 64-bit

走 Docker 时字符串会变成 x86_64-pc-linux-gnu（数据库跑在 Linux 容器里）。只要能打印出版本，就说明 Python 已经连上了，后面所有脚本不用改连接串。

这段代码里有三个东西值得停一下看，它们会在后面每个 Python 示例里出现：

psycopg.connect() 建立的是一条到服务端的真实网络连接，建立它是有明显成本的（第 10 节会实测出大约 50 毫秒），所以不能在循环里反复建。

conn.cursor() 拿到的游标才是真正执行 SQL 的对象。一个连接上可以开多个游标。

with 语句确保连接和游标一定会被清理掉。不用 with 就得自己记着调 close()，忘了就会漏连接——服务端的连接数是有上限的（默认 100 个），漏多了新的连接就连不上了。

5. psql 必会的八条命令 #

psql 里除了能发 SQL，还有一批以反斜杠开头的快捷命令（第 3.8 节说过，这些不是 SQL）。日常够用的就下面八条。

走 Docker 的读者：下面所有在 Windows 里直接敲的 psql ...，改成 docker exec -it pg-tutorial psql ...（要进提示符）或 docker exec pg-tutorial psql ...（执行一条就退出）。进去之后的 \l、\dt、\d 写法不变。

5.1. 命令速查表 #

命令	作用	什么时候用
\l	列出所有数据库	想知道有哪些库
\c 库名	切换到某个库	连错库了要换过去
\dt	列出当前库的所有表	想知道有哪些表
\d 表名	看某张表的结构	最常用，看字段、类型、索引、约束
\di	列出所有索引	检查索引建没建上
\timing	开关计时显示	想知道一条查询花了多久
\x	开关竖排显示	字段太多横着看不清时
\q	退出	结束

5.2. 看库、看表、看表结构 #

这三条是最常用的组合，实际输出长这样。先看有哪些库：

# -l 等价于连进去之后输入 \l，直接在命令行列出所有数据库
psql -U postgres -l

输出（列比较多，这里只保留关键几列）：

     Name     |  Owner   | Encoding |            Collate
--------------+----------+----------+--------------------------------
 langrag      | postgres | UTF8     | Chinese (Simplified)_China.936
 pgdemo       | postgres | UTF8     | Chinese (Simplified)_China.936
 postgres     | postgres | UTF8     | Chinese (Simplified)_China.936
 template0    | postgres | UTF8     | Chinese (Simplified)_China.936
 template1    | postgres | UTF8     | Chinese (Simplified)_China.936

这里有两处要解释。 template0 和 template1 是模板库，新建数据库时 PostgreSQL 会拿它们当模子复制一份，你不要动它们。Collate 那一列是「文本怎么比大小」的规则，本机是中文规则——它会直接影响 ORDER BY 中文时的结果，第 12.1 节专门讲这个。

再看某个库里有哪些表：

# -d 指定库，-c 表示「执行这一条命令然后退出」，适合写在脚本里
# \dt 是 psql 命令，所以要用引号包起来避免被命令行吃掉反斜杠
psql -U postgres -d pgdemo -c "\dt"

输出：

                  List of tables
 Schema |         Name          | Type  |  Owner
--------+-----------------------+-------+----------
 public | articles              | table | postgres
 public | events                | table | postgres
 public | users                 | table | postgres

最后看一张表的详细结构，这条是排查问题时用得最多的：

# \d 加表名，显示这张表的所有字段、类型、约束和索引
psql -U postgres -d pgdemo -c "\d users"

输出：

                                    Table "public.users"
   Column   |           Type           | Collation | Nullable |           Default
------------+--------------------------+-----------+----------+------------------------------
 id         | bigint                   |           | not null | generated always as identity
 name       | text                     |           | not null |
 email      | text                     |           | not null |
 age        | integer                  |           |          |
 created_at | timestamp with time zone |           | not null | now()
Indexes:
    "users_pkey" PRIMARY KEY, btree (id)
    "users_email_key" UNIQUE CONSTRAINT, btree (email)
Check constraints:
    "users_age_check" CHECK (age >= 0)

这份输出把第 3.7 节讲的四种约束全都体现出来了，值得对着看一遍：id 是主键（users_pkey）；email 有唯一约束（users_email_key）；name 和 email 都是 not null；age 上有个检查约束要求非负。另外注意主键和唯一约束下面都标着 btree——它们会自动带一个索引，这也是为什么按 id 或 email 查总是很快。

5.3. 中文 SQL 要写进文件 #

如果你的 SQL 里有中文（中文表名、中文条件值、中文别名），不要直接写在 -c 后面。命令行传参会经过几层编码转换，中文很容易被搅坏：

# 反面例子：SQL 里带中文别名，直接用 -c 传
psql -U postgres -d pgdemo -c "SELECT count(*) AS 行数 FROM users;"

实际报错：

错误:  无效的 "UTF8" 编码字节顺序: 0xd0 0xd0

正确做法是把 SQL 写进一个 UTF-8 编码的 .sql 文件，用 -f 执行。先建文件 demo.sql：

-- 用文件传 SQL，中文不会被命令行编码搅乱
-- 中文别名没问题
SELECT count(*) AS 行数 FROM users;

-- 中文条件值也没问题
SELECT name, age FROM users WHERE name = '张三';

然后执行它：

# 先切 UTF-8 代码页，让输出里的中文能正常显示
chcp 65001 >nul
# -f 表示「执行这个文件里的所有 SQL」
psql -U postgres -d pgdemo -f demo.sql

输出正常了：

 行数
    3
(1 row)

 name | age
------+-----
 张三 |  30
(1 row)

这条经验适用范围比看起来广：凡是要执行的 SQL 超过一两行，或者带中文，都建议写成 .sql 文件。文件还能存进 Git 做版本管理，比命令行历史可靠得多。

6. 建表与增删改查 #

这一节只用 SQL，不涉及 Python。建议在 psql 里边看边敲，每条语句执行完就 SELECT 一下看看效果。

6.1. 建表：先想清楚三件事 #

建表就是告诉数据库「我要存的这类东西有哪些属性，每个属性是什么类型」。动手之前先回答三个问题，能避免后面反复改表：

1. 每行怎么唯一标识？ 绝大多数情况答案是「加一个自增的 id 当主键」。
2. 哪些字段必须有值？ 这些加 NOT NULL。
3. 哪些字段的值有范围限制？ 这些加 CHECK 或 UNIQUE。

先建练习用的数据库。注意这条要在别的库里执行（比如自带的 postgres 库），因为你不能在一个库里创建它自己：

-- 创建一个新数据库，本文后面所有表都建在这里
-- 库名只能用字母数字下划线，别用中文和横线
CREATE DATABASE pgdemo;

然后切换过去：

-- \c 是 psql 命令，切换当前连接的数据库
\c pgdemo

现在建第一张表：

-- IF EXISTS 让「表本来就不存在」时也不报错，这样这段脚本可以反复执行
DROP TABLE IF EXISTS users;

-- CREATE TABLE 后面跟表名，括号里每行定义一个字段
CREATE TABLE users (
    -- 自增主键：GENERATED ALWAYS AS IDENTITY 是标准 SQL 的写法，由数据库自动填值
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    -- 姓名：TEXT 是变长字符串，NOT NULL 表示必填
    name        TEXT NOT NULL,
    -- 邮箱：UNIQUE 保证不重复，NOT NULL 保证必填
    email       TEXT UNIQUE NOT NULL,
    -- 年龄：INT 是整数，CHECK 限制它不能是负数；没写 NOT NULL 所以可以留空
    age         INT CHECK (age >= 0),
    -- 创建时间：TIMESTAMPTZ 是带时区的时间戳，DEFAULT now() 表示不填就用当前时间
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

这五个字段的写法是一个可以直接套用的模板。 几乎每张业务表都会有「自增 id + 若干业务字段 + 创建时间」这个骨架，你要改的只是中间那几个业务字段。

6.2. 插入数据 #

INSERT 的基本形式是「往哪张表的哪几列，插入什么值」：

-- INSERT INTO 表名 (列1, 列2, ...) VALUES (值1, 值2, ...)
-- 注意没有列出 id 和 created_at，它们会自动填
INSERT INTO users (name, email, age)
VALUES ('张三', 'zhang@example.com', 30);

一次插多行，只要多写几组括号，用逗号隔开。这样比执行三次单条 INSERT 快得多，因为只往服务端跑了一趟：

-- 多行插入：VALUES 后面跟多组括号，用逗号分隔
INSERT INTO users (name, email, age)
VALUES
    ('李四', 'li@example.com', 25),
    ('王五', 'wang@example.com', 41),
    ('赵六', 'zhao@example.com', 33);

插入时经常需要知道「刚生成的 id 是多少」。用 RETURNING 让 INSERT 顺便把值返回，省掉一次额外查询：

-- RETURNING 后面写想拿回来的列，这里要 id 和自动生成的时间
INSERT INTO users (name, email, age)
VALUES ('孙七', 'sun@example.com', 28)
RETURNING id, created_at;

输出：

 id |          created_at
----+-------------------------------
  5 | 2026-08-15 01:16:42.123456+08

6.3. 查询数据 #

SELECT 是用得最多的语句。它的完整结构是固定的几段，按这个顺序写就不会错：

SELECT   要哪几列
FROM     从哪张表
WHERE    只要满足什么条件的行
GROUP BY 按什么分组
ORDER BY 按什么排序
LIMIT    最多返回几行

最基础的查询：

-- 只取需要的列，不要习惯性写 SELECT *
-- 列多的时候 SELECT * 会白传很多数据，也让人看不出你到底用了哪些字段
SELECT id, name, age FROM users ORDER BY id;

加条件过滤：

-- WHERE 后面是条件，支持 > < = >= <= <> 以及 AND OR NOT
-- ORDER BY 加 DESC 是从大到小，默认是从小到大（ASC）
SELECT name, age FROM users WHERE age > 28 ORDER BY age DESC;

模糊匹配用 LIKE，% 表示「任意多个字符」：

-- LIKE '%example.com' 表示以 example.com 结尾
-- 想不区分大小写就用 ILIKE（PostgreSQL 特有，MySQL 没有）
SELECT name, email FROM users WHERE email LIKE '%example.com';

统计和分组：

-- count(*) 数行数，是最常用的聚合函数
-- 其他常用的还有 sum() avg() max() min()
SELECT count(*) AS 总人数, avg(age) AS 平均年龄 FROM users;

-- GROUP BY 把行按某个规则分堆，然后对每堆分别聚合
-- CASE WHEN ... THEN ... ELSE ... END 是 SQL 里的 if-else
SELECT
    CASE WHEN age < 30 THEN '30 岁以下' ELSE '30 岁及以上' END AS 年龄段,
    count(*) AS 人数
FROM users
-- GROUP BY 1 表示「按 SELECT 里的第 1 列分组」，省得把长表达式再抄一遍
GROUP BY 1
ORDER BY 1;

分页用 LIMIT 配 OFFSET：

-- LIMIT 10 只要 10 行，OFFSET 20 跳过前 20 行，合起来就是「第 3 页」
-- 分页一定要配 ORDER BY，否则每次返回的顺序可能不一样
SELECT name, age FROM users ORDER BY id LIMIT 10 OFFSET 20;

6.4. 更新与删除 #

这两个操作有个共同的高危点：忘了写 WHERE 会作用到整张表。

-- UPDATE 表名 SET 列 = 新值 WHERE 条件
-- 这里给所有 30 岁以下的人加一岁
UPDATE users SET age = age + 1 WHERE age < 30;

-- 危险示范：没有 WHERE，全表所有人的年龄都会被改成 99
-- UPDATE users SET age = 99;

删除同理：

-- DELETE FROM 表名 WHERE 条件
DELETE FROM users WHERE email = 'sun@example.com';

-- 危险示范：没有 WHERE，整张表的数据都会被删光
-- DELETE FROM users;

一个能救命的习惯：执行 UPDATE 或 DELETE 之前，先把同样的 WHERE 用 SELECT 跑一遍，确认命中的正是你想改的那些行：

-- 第一步：先用 SELECT 确认命中范围
SELECT * FROM users WHERE age < 30;

-- 第二步：确认没问题了，把 SELECT * 换成 UPDATE ... SET
UPDATE users SET age = age + 1 WHERE age < 30;

如果已经手抖执行了，而且还没提交，ROLLBACK 能救回来：

-- 撤销当前事务里的所有改动
ROLLBACK;

但如果已经 COMMIT 了，就只能靠备份恢复了。

6.5. 完整脚本 #

把这一节的内容串成一个可以反复执行的 .sql 文件。存成 basic.sql，用 psql -U postgres -d pgdemo -f basic.sql 运行：

-- ============ 建表 ============
-- 先删旧表，让这个文件可以反复执行
DROP TABLE IF EXISTS users;

-- 建表，把四种约束都用上
CREATE TABLE users (
    -- 自增主键
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    -- 必填的姓名
    name        TEXT NOT NULL,
    -- 必填且不重复的邮箱
    email       TEXT UNIQUE NOT NULL,
    -- 可以留空，但不能是负数
    age         INT CHECK (age >= 0),
    -- 不填就自动记当前时间
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- ============ 插入 ============
-- 一次插三行
INSERT INTO users (name, email, age) VALUES
    ('张三', 'zhang@example.com', 30),
    ('李四', 'li@example.com', 25),
    ('王五', 'wang@example.com', 41);

-- 插一行并把生成的 id 返回
INSERT INTO users (name, email, age)
VALUES ('赵六', 'zhao@example.com', 33)
RETURNING id;

-- ============ 查询 ============
-- 全部数据
SELECT id, name, age FROM users ORDER BY id;

-- 条件过滤
SELECT name, age FROM users WHERE age > 28 ORDER BY age DESC;

-- 聚合统计
SELECT count(*) AS 总人数 FROM users;

-- 分组统计
SELECT
    CASE WHEN age < 30 THEN '30 岁以下' ELSE '30 岁及以上' END AS 年龄段,
    count(*) AS 人数
FROM users
GROUP BY 1
ORDER BY 1;

-- ============ 更新与删除 ============
-- 改之前先确认命中范围
SELECT id, name, age FROM users WHERE age < 30;

-- 确认后再改
UPDATE users SET age = age + 1 WHERE age < 30;

-- 删一行
DELETE FROM users WHERE email = 'zhao@example.com';

-- 看最终结果
SELECT id, name, age FROM users ORDER BY id;

7. 用 Python 操作数据库 #

实际项目里绝大多数 SQL 是由程序发出的，不是手敲的。这一节讲 Python 侧的正确写法，其中 7.2 和 7.4 是两个必须掌握否则一定出事的点。

7.1. psycopg 的三层对象 #

psycopg 里有三个对象，层层嵌套，各管一段：

连接 Connection ← 一条到服务端的通道，管事务（commit / rollback）
   └── 游标 Cursor ← 执行 SQL、取结果
         └── 结果 ← fetchone() 取一行，fetchall() 取全部

标准写法是两层 with 嵌套，这样连接和游标都会被自动清理：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

# 第一层 with：建立连接，块结束时自动关闭
with psycopg.connect(DSN) as conn:
    # 第二层 with：拿游标，块结束时自动关闭
    with conn.cursor() as cur:
        # 用游标执行 SQL
        cur.execute("SELECT count(*) FROM users")
        # 取一行结果；结果是元组，用 [0] 取第一列
        print("用户数:", cur.fetchone()[0])

取结果有三个方法，按你要几行来选：

方法	返回	什么时候用
fetchone()	一个元组，没有了就返回 None	count(*) 这种只有一行的结果
fetchall()	元组的列表	确定结果不多，一次全拿走
直接 for row in cur	一行一行迭代	结果可能很大，不想全塞进内存

7.2. 参数化查询：不只是防注入 #

这一节是本文最重要的一条规矩：往 SQL 里塞值，必须用参数化，永远不要用字符串拼接。

正确写法是用 %s 占位，真实值作为第二个参数传进去：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 要查的邮箱，假设它来自用户输入
        email = "zhang@example.com"
        # %s 是占位符，值放在第二个参数里，用元组包着
        # 即使只有一个值，也要写成 (email,) 这样带逗号的单元素元组
        cur.execute("SELECT name, age FROM users WHERE email = %s", (email,))
        print(cur.fetchone())

为什么不能拼字符串？下面这段把两种写法放在一起对比，用一个精心构造的输入试探它们：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 一个恶意输入：末尾的 ' OR '1'='1 想把 WHERE 条件变成「永远为真」
        evil = "x@example.com' OR '1'='1"

        # 写法一：参数化。驱动会把整个字符串当成一个「值」，不会当成 SQL 来解释
        cur.execute("SELECT count(*) FROM users WHERE email = %s", (evil,))
        print("参数化，匹配行数:", cur.fetchone()[0])

        # 写法二：拼字符串。用户的输入直接变成了 SQL 的一部分
        bad_sql = f"SELECT count(*) FROM users WHERE email = '{evil}'"
        cur.execute(bad_sql)
        print("拼字符串，匹配行数:", cur.fetchone()[0])
        # 把拼出来的 SQL 打印出来，一眼就能看出问题
        print("拼出来的 SQL:", bad_sql)

实测输出：

参数化，匹配行数: 0
拼字符串，匹配行数: 3
拼出来的 SQL: SELECT count(*) FROM users WHERE email = 'x@example.com' OR '1'='1'

这三行输出说明了一切。 参数化版本老老实实地去找「邮箱等于 `x@example.com' OR '1'='1这个字符串」的人，找不到，返回 0——完全符合预期。拼字符串版本则把用户输入当成了 SQL 代码，最终执行的条件变成了「邮箱等于某值 **或者**'1'='1'`」，而后半句永远成立，于是整张表 3 行全部返回。

把这个手法用在登录接口上就是「不用密码也能登进去」，用在 DELETE 上就是「删光整张表」。这类攻击叫 SQL 注入，是最古老也依然最常见的漏洞之一。

除了安全，参数化还有两个实际好处，这也是为什么就算输入完全可信也该这么写：

1. 不用操心引号和转义。 名字里带单引号的人（比如 O'Brien）用拼接会直接让 SQL 语法出错，参数化则毫无问题。
2. 类型自动转换。 Python 的 datetime、list、dict 传进去会被转成对应的 PostgreSQL 类型，不用自己拼格式字符串。

一个例外要知道： 占位符只能用在「值」的位置，不能用来传表名或列名。SELECT * FROM %s 是不行的。表名确实需要动态拼的话，用 psycopg.sql.Identifier 这个专门的工具，不要手拼。

7.3. 事务：commit 与 rollback #

psycopg 默认不是自动提交的。这意味着你 INSERT 完如果不 commit，数据不会真正落库。下面这段用两个连接演示这个现象：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

# 开第一个连接，用它来写
c1 = psycopg.connect(DSN)
# 插一行，但故意不提交
c1.cursor().execute("INSERT INTO users (name, email, age) VALUES ('没提交', 'nc@example.com', 9)")

# 开第二个连接，模拟「另一个程序」
c2 = psycopg.connect(DSN)
cur2 = c2.cursor()
# 从第二个连接查刚插入的那行
cur2.execute("SELECT count(*) FROM users WHERE email = 'nc@example.com'")
print("另一个连接看到的行数:", cur2.fetchone()[0])

# 现在第一个连接提交
c1.commit()
# 第二个连接再查一次
cur2.execute("SELECT count(*) FROM users WHERE email = 'nc@example.com'")
print("commit 之后再看:", cur2.fetchone()[0])

# 清理掉这行测试数据
c1.cursor().execute("DELETE FROM users WHERE email = 'nc@example.com'")
c1.commit()
# 手动开的连接要手动关
c1.close()
c2.close()

实测输出：

另一个连接看到的行数: 0
commit 之后再看: 1

这个 0 就是「忘了 commit」这个坑的真实样子。 你的程序里 INSERT 没报错、rowcount 也是 1，一切看起来正常，但换个连接（或者程序重启后）数据就是不在。所以规矩很简单：每一批写操作之后，都要有一个 commit。

想撤销就用 rollback：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 先看看现在多少行
        cur.execute("SELECT count(*) FROM users")
        print("开始时:", cur.fetchone()[0], "行")

        # 插一行，不提交
        cur.execute("INSERT INTO users (name, email, age) VALUES ('临时', 'tmp@example.com', 1)")
        cur.execute("SELECT count(*) FROM users")
        # 在同一个连接里，能看到自己未提交的改动
        print("插入后（本连接可见）:", cur.fetchone()[0], "行")

        # rollback 把这个事务里的所有改动一起撤销
        conn.rollback()
        cur.execute("SELECT count(*) FROM users")
        print("rollback 后:", cur.fetchone()[0], "行")

实测输出：

开始时: 3 行
插入后（本连接可见）: 4 行
rollback 后: 3 行

注意中间那行：自己未提交的改动，自己是看得见的（4 行），只有别的连接看不见。理解这一点，调试时就不会因为「我这边明明查到了」而困惑。

如果你的场景确实每条都想立即生效（比如建库、建表这类管理操作），可以开自动提交：

# 驱动
import psycopg

# 连接串：注意连的是自带的 postgres 库
ADMIN_DSN = "postgresql://postgres:postgres@localhost:5432/postgres"

# autocommit=True 表示每条语句执行完立刻生效，不用手动 commit
# CREATE DATABASE 这类语句必须在自动提交模式下执行，否则会报错
with psycopg.connect(ADMIN_DSN, autocommit=True) as conn:
    with conn.cursor() as cur:
        # 先查系统表看库是不是已经存在
        cur.execute("SELECT 1 FROM pg_database WHERE datname = 'pgdemo'")
        # 不存在才创建
        if not cur.fetchone():
            # 注意数据库名不能用参数化占位符，只能拼在语句里
            cur.execute("CREATE DATABASE pgdemo")
            print("已创建 pgdemo")
        else:
            print("pgdemo 已存在")

7.4. 出错后必须 rollback #

这个坑非常隐蔽，几乎每个 psycopg 初学者都会中一次：一个事务里只要有一条语句出错，这个事务就进入「失败」状态，之后所有语句都会被拒绝——哪怕语句本身完全正确。

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 第一步：故意制造一个错误
        try:
            cur.execute("SELECT 1 / 0")
        except psycopg.errors.DivisionByZero:
            print("先制造一个除零错误")

        # 第二步：不 rollback，直接执行一条完全正常的语句
        try:
            cur.execute("SELECT 1")
        except psycopg.errors.InFailedSqlTransaction as e:
            # 它也会失败，报错说「当前事务被终止」
            print("正常语句也被拒绝:", str(e).splitlines()[0])

        # 第三步：rollback 之后，连接恢复可用
        conn.rollback()
        cur.execute("SELECT 1")
        print("rollback 后恢复正常:", cur.fetchone()[0])

实测输出：

先制造一个除零错误
正常语句也被拒绝: 当前事务被终止, 事务块结束之前的查询被忽略
rollback 后恢复正常: 1

为什么这个坑特别难查？ 因为你看到的报错（InFailedSqlTransaction）跟真正的错因（除零）毫无关系，而且真正的错误可能发生在几十行代码之前，甚至在另一个函数里。你会盯着一条明明没问题的 SELECT 1 百思不得其解。

记住这条规则就行：捕获了数据库异常之后，一定要 conn.rollback()。 写成模板：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        try:
            # 这里放可能失败的写操作
            cur.execute("INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
                        ("重复", "zhang@example.com", 20))
            # 成功了就提交
            conn.commit()
        except psycopg.errors.UniqueViolation:
            # 失败了先回滚，让连接恢复可用，再做你的错误处理
            conn.rollback()
            print("这个邮箱已经存在了")

7.5. 批量写入：executemany 与 COPY #

要插很多行时，写法的选择会带来数量级的性能差异。有三种写法，从慢到快：

写法	原理	适用规模
循环里逐条 execute	每行单独发一次，N 行 N 次往返	别用
executemany	驱动把多组参数打包发送	几百到几千行
COPY	PostgreSQL 专用的批量导入通道	上万行

差距有多大值得实测一次。下面这个脚本用同样的 5000 行数据，把三种写法各跑一遍。存成 pg_bulk.py，可以反复运行：

"""三种批量写入方式的耗时对比：逐条 execute、executemany、COPY。

可反复运行。用自己独立的 bulk_demo 表，不影响其他示例。
"""

# 计时
import time

# 驱动
import psycopg

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"
# 每种方式各插多少行
N = 5000


# 每次测试前把表重建一次，保证三种方式的起点一样
def prepare(cur: psycopg.Cursor) -> None:
    """重建测试表。"""
    # 先删掉旧表，这样脚本可以反复运行
    cur.execute("DROP TABLE IF EXISTS bulk_demo")
    # 建一张最简单的表，只有自增 id 和两个业务字段
    cur.execute("""
        CREATE TABLE bulk_demo (
            id    BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
            name  TEXT NOT NULL,
            score INT  NOT NULL
        )
    """)


# 先在内存里造好要插入的数据，避免把造数据的时间算进对比
rows = [(f"用户{i}", i % 100) for i in range(N)]

# 建立连接和游标
with psycopg.connect(DSN) as conn, conn.cursor() as cur:
    # ---------- 方式一：循环里逐条 execute（最慢） ----------
    # 重建表
    prepare(cur)
    conn.commit()
    # 开始计时
    t0 = time.perf_counter()
    # 每一行都单独发一条 INSERT，5000 行就是 5000 次往返
    for name, score in rows:
        cur.execute("INSERT INTO bulk_demo (name, score) VALUES (%s, %s)", (name, score))
    # 全部插完后一次提交
    conn.commit()
    # 记下耗时
    t_loop = time.perf_counter() - t0
    # 确认真的插进去了
    cur.execute("SELECT count(*) FROM bulk_demo")
    print(f"逐条 execute : {t_loop:.3f} 秒，{cur.fetchone()[0]} 行")

    # ---------- 方式二：executemany（较快） ----------
    # 重建表
    prepare(cur)
    conn.commit()
    # 开始计时
    t0 = time.perf_counter()
    # executemany 的第二个参数是「参数组的列表」，驱动会把它们打包发送
    cur.executemany("INSERT INTO bulk_demo (name, score) VALUES (%s, %s)", rows)
    # 提交
    conn.commit()
    # 记下耗时
    t_many = time.perf_counter() - t0
    # 确认行数
    cur.execute("SELECT count(*) FROM bulk_demo")
    print(f"executemany  : {t_many:.3f} 秒，{cur.fetchone()[0]} 行")

    # ---------- 方式三：COPY（最快） ----------
    # 重建表
    prepare(cur)
    conn.commit()
    # 开始计时
    t0 = time.perf_counter()
    # cur.copy() 打开一个 COPY 流；FROM STDIN 表示数据由客户端流式送过来
    with cur.copy("COPY bulk_demo (name, score) FROM STDIN") as cp:
        # 逐行写进流里，不用拼 SQL 也不用解析
        for r in rows:
            cp.write_row(r)
    # 提交
    conn.commit()
    # 记下耗时
    t_copy = time.perf_counter() - t0
    # 确认行数
    cur.execute("SELECT count(*) FROM bulk_demo")
    print(f"COPY         : {t_copy:.3f} 秒，{cur.fetchone()[0]} 行")

    # ---------- 对比 ----------
    # 打一行空行分隔
    print()
    # executemany 相对逐条的提升
    print(f"executemany 比逐条快 {t_loop / t_many:.1f} 倍")
    # COPY 相对逐条的提升
    print(f"COPY 比逐条快 {t_loop / t_copy:.1f} 倍")
    # COPY 相对 executemany 的提升
    print(f"COPY 比 executemany 快 {t_many / t_copy:.1f} 倍")

实测输出：

逐条 execute : 0.342 秒，5000 行
executemany  : 0.087 秒，5000 行
COPY         : 0.011 秒，5000 行

executemany 比逐条快 3.9 倍
COPY 比逐条快 30.0 倍
COPY 比 executemany 快 7.6 倍

结论很清楚：COPY 比逐条 INSERT 快 30 倍。 换算下来，5000 行用 COPY 只要 11 毫秒，平均每行 2.2 微秒。

这里还有一个更重要的提醒：上面这组数字是在本机测的，真实场景差距会更大。 因为逐条写入的开销主要是「往返次数」，本机往返几乎不要时间，而数据库在另一台机器上时，每次往返都要加上网络延迟。假设延迟 1 毫秒，5000 次往返光网络就是 5 秒——而 COPY 只需要一次往返。所以数据库不在本机时，这个差距会从 30 倍拉大到几百倍。

判断标准：几百行以内用 executemany，上万行用 COPY，任何情况下都不要在循环里逐条 execute。

7.6. 完整脚本与输出 #

把这一节的要点串成一个可以反复运行的完整脚本。存成 pg_basic.py，直接 python pg_basic.py 就能跑：

"""PostgreSQL 基础操作全流程：建库 → 建表 → 增删改查 → 事务。

这个脚本可以反复运行，每次都会先删掉旧表再重建。
运行前只要保证：本机 PostgreSQL 已启动，密码是 postgres。
"""

# psycopg 是 PostgreSQL 的 Python 驱动，3.x 版本的包名就叫 psycopg
import psycopg

# 连 postgres 这个「自带库」，专门用来执行建库这类管理操作
ADMIN_DSN = "postgresql://postgres:postgres@localhost:5432/postgres"
# 连我们自己的练习库，后面所有表都建在这里
DEMO_DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"


# 第 1 步：确保练习库存在
def create_database() -> None:
    """确保练习库 pgdemo 存在。"""
    # CREATE DATABASE 不允许在事务里执行，所以必须开 autocommit
    with psycopg.connect(ADMIN_DSN, autocommit=True) as conn:
        # 游标（cursor）是执行 SQL、取结果的把手
        with conn.cursor() as cur:
            # 先查系统表 pg_database，看这个库是不是已经有了
            cur.execute("SELECT 1 FROM pg_database WHERE datname = 'pgdemo'")
            # fetchone() 有结果说明库已存在
            if cur.fetchone():
                # 已存在就打印一句然后直接返回，不重复创建
                print("pgdemo 已存在")
                return
            # 库不存在才创建；注意库名不能用参数化占位符，只能拼在语句里
            cur.execute("CREATE DATABASE pgdemo")
            # 告诉使用者创建成功了
            print("已创建 pgdemo")


# 第 2 步：建表
def create_table() -> None:
    """建一张用户表，顺便把常用约束都用上。"""
    # 连练习库；这里不需要 autocommit，因为后面要手动控制事务
    with psycopg.connect(DEMO_DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # IF EXISTS 让「表本来不存在」也不报错，脚本就能反复跑
            cur.execute("DROP TABLE IF EXISTS users")
            # 建表语句：每个字段一行，逗号分隔
            cur.execute("""
                CREATE TABLE users (
                    -- 自增主键：由数据库自动生成，用 BIGINT 避免将来撞上限
                    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                    -- 姓名：必填
                    name        TEXT NOT NULL,
                    -- 邮箱：必填且不能重复
                    email       TEXT UNIQUE NOT NULL,
                    -- 年龄：可以留空，但不能是负数
                    age         INT CHECK (age >= 0),
                    -- 创建时间：不填就自动记当前时间，带时区
                    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
                )
            """)
            # DDL（建表这类语句）也在事务里，不提交同样会丢
            conn.commit()
            # 确认建好
            print("users 表已建好")


# 第 3 步：插入数据
def insert_rows() -> None:
    """插入数据：单条、多条、以及拿回自增 id。"""
    # 连库
    with psycopg.connect(DEMO_DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # executemany 把同一条语句套用到多组参数上
            cur.executemany(
                # %s 是占位符，真实值由第二个参数提供，永远不要用字符串拼接
                "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
                # 每个元组对应一行，顺序和上面的字段顺序一致
                [
                    ("张三", "zhang@example.com", 30),
                    ("李四", "li@example.com", 25),
                    ("王五", "wang@example.com", 41),
                ],
            )
            # RETURNING 让 INSERT 顺便把生成的值返回，省一次查询
            cur.execute(
                "INSERT INTO users (name, email, age) VALUES (%s, %s, %s) RETURNING id, created_at",
                ("赵六", "zhao@example.com", 33),
            )
            # 取出刚插入那一行的 id 和创建时间
            new_id, created = cur.fetchone()
            # 打印出来确认自增 id 生效了
            print(f"新插入的 id={new_id}，创建时间={created:%Y-%m-%d %H:%M:%S}")
            # 提交，数据才真正落库
            conn.commit()


# 第 4 步：各种查询写法
def query_rows() -> None:
    """几种最常用的查询写法。"""
    # 连库
    with psycopg.connect(DEMO_DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # 只取需要的列，别习惯性写 SELECT *
            cur.execute("SELECT id, name, age FROM users ORDER BY id")
            # fetchall() 把所有行读成列表，每行是一个元组
            print("全部用户:", cur.fetchall())

            # WHERE 过滤 + 参数化传值；单个值也要写成带逗号的元组
            cur.execute("SELECT name, age FROM users WHERE age > %s ORDER BY age DESC", (28,))
            # 打印过滤结果
            print("28 岁以上:", cur.fetchall())

            # fetchone() 取一行，适合 count 这种只有一行的结果
            cur.execute("SELECT count(*) FROM users")
            # [0] 是取这一行的第一列
            print("总人数:", cur.fetchone()[0])

            # 聚合 + 分组：按年龄段统计
            cur.execute("""
                SELECT CASE WHEN age < 30 THEN '30 岁以下' ELSE '30 岁及以上' END AS 分组,
                       count(*) AS 人数
                FROM users
                GROUP BY 1
                ORDER BY 1
            """)
            # 结果是每个分组一行
            print("分组统计:", cur.fetchall())

            # LIMIT 限制返回行数，做分页或抽样时常用
            cur.execute("SELECT name FROM users ORDER BY age DESC LIMIT 2")
            # 只返回两行
            print("年龄最大的两个:", cur.fetchall())


# 第 5 步：更新与删除
def update_and_delete() -> None:
    """更新与删除，重点看 rowcount。"""
    # 连库
    with psycopg.connect(DEMO_DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # UPDATE 一定要带 WHERE，否则整表都会被改
            cur.execute("UPDATE users SET age = age + 1 WHERE age < 30")
            # rowcount 是「影响了多少行」，用它确认是否改到了预期的量
            print("涨了一岁的人数:", cur.rowcount)

            # DELETE 同样必须带 WHERE
            cur.execute("DELETE FROM users WHERE email = %s", ("zhao@example.com",))
            # 同样用 rowcount 确认删掉了几行
            print("删掉的行数:", cur.rowcount)
            # 提交这两个改动
            conn.commit()


# 第 6 步：演示「不提交等于没做」
def transaction_demo() -> None:
    """事务：不提交就等于没做。"""
    # 连库
    with psycopg.connect(DEMO_DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # 先看看现在有多少行
            cur.execute("SELECT count(*) FROM users")
            # 记下基准数字
            print("开始时:", cur.fetchone()[0], "行")

            # 插一行但故意不提交
            cur.execute("INSERT INTO users (name, email, age) VALUES ('临时', 'tmp@example.com', 1)")
            # 再数一次
            cur.execute("SELECT count(*) FROM users")
            # 在同一个连接里能看到自己未提交的改动，所以这里会多一行
            print("插入后（本连接可见）:", cur.fetchone()[0], "行")

            # rollback 把这个事务里所有改动一起撤销
            conn.rollback()
            # 第三次数，应该回到基准数字
            cur.execute("SELECT count(*) FROM users")
            # 证明刚才那行确实没留下
            print("rollback 后:", cur.fetchone()[0], "行")


# 第 7 步：演示约束拦截与出错后必须 rollback
def error_demo() -> None:
    """违反约束会抛什么异常，以及出错后为什么必须 rollback。"""
    # 连库
    with psycopg.connect(DEMO_DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # 邮箱有 UNIQUE 约束，插重复值会被数据库拦下
            try:
                # 这个邮箱前面已经插过了
                cur.execute("INSERT INTO users (name, email, age) VALUES ('重复', 'zhang@example.com', 20)")
            # 唯一约束冲突对应 UniqueViolation
            except psycopg.errors.UniqueViolation as e:
                # 只打第一行，后面几行是重复的细节
                print("UniqueViolation:", str(e).splitlines()[0])
            # 出错后事务进入失败状态，必须 rollback 才能继续用这个连接
            conn.rollback()

            # age 有 CHECK 约束，负数同样插不进去
            try:
                # 年龄给 -5，违反 CHECK (age >= 0)
                cur.execute("INSERT INTO users (name, email, age) VALUES ('负数', 'neg@example.com', -5)")
            # 检查约束冲突对应 CheckViolation
            except psycopg.errors.CheckViolation as e:
                # 打印报错第一行
                print("CheckViolation:", str(e).splitlines()[0])
            # 同样要回滚
            conn.rollback()

            # 下面演示「不 rollback 会怎样」：先制造一个错
            try:
                # 除以零，一定报错
                cur.execute("SELECT 1 / 0")
            # 除零对应 DivisionByZero
            except psycopg.errors.DivisionByZero:
                # 确认第一个错已经发生
                print("先制造一个除零错误")
            # 关键：这里故意不 rollback，直接执行一条完全正常的语句
            try:
                # SELECT 1 语法和语义都没问题
                cur.execute("SELECT 1")
            # 但它会因为事务已失败而被拒绝
            except psycopg.errors.InFailedSqlTransaction as e:
                # 报错说的是「当前事务被终止」，跟真正的错因（除零）完全无关
                print("正常语句也被拒绝:", str(e).splitlines()[0])
            # rollback 之后连接恢复可用
            conn.rollback()
            # 再执行同一条语句，这次成功
            cur.execute("SELECT 1")
            # 证明连接已经恢复
            print("rollback 后恢复正常:", cur.fetchone()[0])


# 只有直接运行这个文件时才执行下面的流程，被 import 时不会触发
if __name__ == "__main__":
    # 按顺序把上面每一步跑一遍
    for step in (
        # 建库
        create_database,
        # 建表
        create_table,
        # 插数据
        insert_rows,
        # 查数据
        query_rows,
        # 改和删
        update_and_delete,
        # 事务演示
        transaction_demo,
        # 报错演示
        error_demo,
    ):
        # 每一步之前打一行分隔，方便对照输出
        print(f"\n----- {step.__name__} -----")
        # 调用这一步的函数
        step()

实测输出：

----- create_database -----
pgdemo 已存在

----- create_table -----
users 表已建好

----- insert_rows -----
新插入的 id=4，创建时间=2026-08-15 01:16:42

----- query_rows -----
全部用户: [(1, '张三', 30), (2, '李四', 25), (3, '王五', 41), (4, '赵六', 33)]
28 岁以上: [('王五', 41), ('赵六', 33), ('张三', 30)]
总人数: 4
分组统计: [('30 岁及以上', 3), ('30 岁以下', 1)]
年龄最大的两个: [('王五',), ('赵六',)]

----- update_and_delete -----
涨了一岁的人数: 1
删掉的行数: 1

----- transaction_demo -----
开始时: 3 行
插入后（本连接可见）: 4 行
rollback 后: 3 行

----- error_demo -----
UniqueViolation: 重复键违反唯一约束"users_email_key"
CheckViolation: 关系 "users" 的新列违反了检查约束 "users_age_check"
先制造一个除零错误
正常语句也被拒绝: 当前事务被终止, 事务块结束之前的查询被忽略
rollback 后恢复正常: 1

注意 id=4 这个细节。 表是重新建的，但自增 id 从 1 开始给了前三行，第四行是 4——说明 executemany 插的三行也各自拿到了 id。另外 分组统计 里「30 岁以下」只有 1 人，因为 25 岁的李四在 update_and_delete 之前统计的。

8. 常用数据类型 #

PostgreSQL 支持的类型有几十种，但日常业务用到的就那么几个。这一节只讲会真正用到的，冷门类型（几何、网络地址、范围类型等）遇到再查文档就行。

8.1. 类型速查表 #

类型	存什么	对应 Python 类型	备注
TEXT	字符串	str	首选，不用纠结长度
INT	整数	int	范围约 ±21 亿
BIGINT	大整数	int	id 用这个更保险
NUMERIC(p,s)	精确小数	Decimal	钱必须用这个
DOUBLE PRECISION	浮点数	float	科学计算用，钱别用
BOOLEAN	真假	bool	值是 true / false
TIMESTAMPTZ	带时区的时间	datetime	时间首选这个
DATE	只有日期	date	生日这类
JSONB	JSON 数据	dict / list	结构不固定的数据
TEXT[]	字符串数组	list	标签这类
UUID	通用唯一标识	UUID	分布式系统里当主键

常问的两个问题，这里直接给答案：

「VARCHAR(50) 和 TEXT 用哪个？」——在 PostgreSQL 里一律用 TEXT。和 MySQL 不同，PostgreSQL 里这两者性能完全一样，VARCHAR(n) 只是多加了个长度检查。真要限制长度，用 CHECK (length(name) <= 50) 更清晰，而且以后改限制不用改列类型。

「CHAR(n) 呢？」——永远不要用。它会把字符串用空格补齐到固定长度，几乎总是给你带来意外。

8.2. 自增主键：用 IDENTITY 而不是 SERIAL #

网上很多老教程用 SERIAL，但现在推荐用 IDENTITY：

-- 老写法：SERIAL。它其实是个语法糖，背后会偷偷建一个序列对象
-- 问题是这个序列的权限、归属管理起来很别扭，而且不是标准 SQL
CREATE TABLE old_way (
    id SERIAL PRIMARY KEY,
    name TEXT
);

-- 推荐写法：IDENTITY。这是标准 SQL，PostgreSQL 10 以后支持
-- GENERATED ALWAYS 表示「这一列永远由数据库生成」，你手动插值会被拒绝，更安全
CREATE TABLE new_way (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name TEXT
);

为什么用 BIGINT 而不是 INT？ INT 的上限是约 21 亿。听起来很多，但如果这张表有频繁的插入和删除，id 是一直往上涨不会回收的，撞到上限的项目真实存在过。改类型要锁表重写，代价很大。开头就用 BIGINT，多出的 4 个字节完全不值得省。

如果确实需要手动指定 id（比如数据迁移），把 ALWAYS 换成 BY DEFAULT：

-- GENERATED BY DEFAULT 表示「你不给我就自动生成，你给了就用你的」
CREATE TABLE flexible (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    name TEXT
);

8.3. 时间：一律用 TIMESTAMPTZ #

PostgreSQL 有两个时间戳类型，名字只差三个字母，但行为差别很大：

类型	全名	存不存时区	结果
TIMESTAMP	timestamp without time zone	不存	换个时区看，时间还是那个数，含义变了
TIMESTAMPTZ	timestamp with time zone	存	换个时区看，自动转成当地时间，含义不变

结论很简单：默认一律用 TIMESTAMPTZ。 理由是即使你现在只有国内用户，一旦服务器在国外、或者用户出国、或者要跟第三方对接，不带时区的时间就会变成一笔糊涂账，而修数据比一开始选对类型难一百倍。

-- 建表时用 TIMESTAMPTZ，配 DEFAULT now() 自动记录创建时间
CREATE TABLE logs (
    id         BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    message    TEXT NOT NULL,
    -- now() 返回当前时间（带时区），NOT NULL 保证这一列一定有值
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

读出来在 Python 里是带时区的 datetime，可以直接比较和运算：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 取出创建时间
        cur.execute("SELECT created_at FROM users ORDER BY id LIMIT 1")
        ts = cur.fetchone()[0]
        # 打印值和类型，可以看到它带着时区信息
        print("值:", ts)
        print("类型:", type(ts).__name__)
        print("时区:", ts.tzinfo)

输出：

值: 2026-08-15 01:17:30.580291+08:00
类型: datetime
时区: Asia/Shanghai

8.4. 小数：钱不要用 float #

这是个会造成真实资金损失的坑。浮点数（float / DOUBLE PRECISION）在二进制里存不下某些十进制小数，会有微小误差：

# 这是 Python 自带的现象，跟数据库无关，但同一个原理
# 0.1 和 0.2 在二进制浮点里都存不精确，加起来就露馅了
print(0.1 + 0.2)
print(0.1 + 0.2 == 0.3)

输出：

0.30000000000000004
False

一分钱的误差乘上几百万笔交易就是大问题。存钱、存需要精确的小数，用 NUMERIC：

-- NUMERIC(10, 2) 表示总共 10 位数字，其中 2 位是小数
-- 也就是最大能存 99999999.99
CREATE TABLE orders (
    id     BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    -- 金额用 NUMERIC，精确不会有误差
    amount NUMERIC(10, 2) NOT NULL CHECK (amount >= 0)
);

读到 Python 里是 Decimal 类型，同样是精确的：

# Decimal 是 Python 标准库里的精确小数类型
from decimal import Decimal

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 建一张临时表
        cur.execute("DROP TABLE IF EXISTS money_test")
        cur.execute("CREATE TABLE money_test (amount NUMERIC(10, 2))")
        # 插两个值，注意用 Decimal 而不是 float 传参
        cur.executemany("INSERT INTO money_test (amount) VALUES (%s)",
                        [(Decimal("0.1"),), (Decimal("0.2"),)])
        conn.commit()
        # 在数据库里求和
        cur.execute("SELECT sum(amount) FROM money_test")
        total = cur.fetchone()[0]
        # 精确等于 0.3，没有误差
        print("求和结果:", total, "| 类型:", type(total).__name__)
        print("等于 0.3 吗:", total == Decimal("0.3"))

输出：

求和结果: 0.30 | 类型: Decimal
等于 0.3 吗: True

8.5. JSONB：半结构化数据的解法 #

有时候你要存的数据结构不固定。比如商品的属性：手机有屏幕尺寸和内存，衣服有尺码和颜色，图书有作者和 ISBN。硬要给每种属性建一列，表会宽得没法维护。

这时候用 JSONB：一列里存一个完整的 JSON 对象。

-- meta 这一列可以存任意结构的 JSON
-- DEFAULT '{}'::jsonb 表示不填就存一个空对象，避免出现 NULL 要额外判断
CREATE TABLE products (
    id    BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name  TEXT  NOT NULL,
    meta  JSONB NOT NULL DEFAULT '{}'::jsonb
);

为什么是 JSONB 而不是 JSON？ PostgreSQL 有两个 JSON 类型：

	JSON	JSONB
存储方式	原样存文本	转成二进制格式存
写入速度	稍快	稍慢（要解析）
查询速度	慢（每次都要重新解析）	快
能不能建索引	不能	能（GIN 索引）
保留键的顺序	保留	不保留
保留重复的键	保留	只留最后一个

除了极少数需要原样保留 JSON 文本的场景，一律用 JSONB。 它唯一的「代价」是不保留键顺序，而 JSON 对象的键顺序本来就不该被依赖。

8.6. JSONB 操作符速查 #

JSONB 的操作符看起来有点像天书，但常用的只有六个。这张表配合下面的实测输出看，很快就能记住：

操作符	干什么	例子	返回
->	取值，结果还是 JSON	meta -> 'tags'	["制度","入职"]
->>	取值，结果是文本	meta ->> 'author'	张三
#>>	按路径取深层的值（文本）	meta #>> '{topics,0}'	SQL
@>	左边是否包含右边	meta @> '{"dept":"hr"}'	true / false
?	某个键（或数组元素）是否存在	meta ? 'level'	true / false
`\	\	`	合并两个 JSONB	`meta \	\	'{"lang":"zh"}'`	合并后的对象

-> 和 ->> 的区别是最常被搞混的一点，一句话记住：多一个 > 就多剥一层，变成纯文本。 需要拿去和字符串比较、或者直接显示给人看时用 ->>；还要继续往下取的时候用 ->。

还有一个高频陷阱：JSON 里的数字取出来也是文本，比大小之前必须转类型：

-- 错误写法：->> 拿到的是文本，'9' > '10' 在文本比较里是成立的
-- SELECT * FROM products WHERE meta ->> 'price' > '10';

-- 正确写法：先用 ::int 或 ::numeric 转成数字再比
SELECT * FROM products WHERE (meta ->> 'price')::numeric > 10;

8.7. 数组与枚举 #

数组适合存「同一类的多个值」，比如标签：

-- TEXT[] 表示「字符串数组」，方括号就是数组的标志
CREATE TABLE posts (
    id   BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    title TEXT NOT NULL,
    -- 标签数组
    tags TEXT[]
);

-- 插入数组用 ARRAY[...] 或者 '{...}' 两种写法都行
INSERT INTO posts (title, tags) VALUES ('入门指南', ARRAY['数据库', '入门']);

-- @> 判断数组是否包含某个元素
SELECT title FROM posts WHERE tags @> ARRAY['入门'];

-- unnest 把数组展开成多行，做标签统计时常用
SELECT unnest(tags) AS tag, count(*) FROM posts GROUP BY 1 ORDER BY 2 DESC;

枚举适合「只能取固定几个值」的字段，比状态码用字符串更安全：

-- 先定义类型，列出所有允许的值
-- IF EXISTS 让这段可以反复执行
DROP TYPE IF EXISTS article_status;
CREATE TYPE article_status AS ENUM ('draft', 'published', 'archived');

-- 然后就能像内置类型一样用它
CREATE TABLE articles2 (
    id     BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    title  TEXT NOT NULL,
    -- 这一列只能是上面三个值之一，写别的直接报错
    status article_status NOT NULL DEFAULT 'draft'
);

枚举的取舍要提一下： 它的好处是数据库层面就挡住了非法值；坏处是加一个新值需要改类型定义（ALTER TYPE ... ADD VALUE），比改一个 CHECK 约束麻烦。如果这几个值将来可能频繁增减，用 TEXT 加 CHECK 反而更灵活。

8.8. Python 类型与 PostgreSQL 类型的对应 #

psycopg 会自动做双向转换，不用手写任何格式字符串。这是实测的对应关系：

PostgreSQL 类型	读到 Python 是	写入时传什么
TEXT	str	str
INT / BIGINT	int	int
NUMERIC	Decimal	Decimal（别传 float）
BOOLEAN	bool	bool
TIMESTAMPTZ	datetime（带时区）	datetime
TEXT[]	list	list
JSONB	dict / list	Jsonb(...) 包一层，或传 JSON 字符串
枚举	str	str

只有 JSONB 需要特别说明。 直接传 Python 字典 psycopg 不知道你是想存 JSON 还是存别的复合类型，所以要用 Jsonb 包一下：

# 驱动
import psycopg
# Jsonb 是 psycopg 提供的包装类，告诉驱动「这个字典要当 JSONB 存」
from psycopg.types.json import Jsonb

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 建表
        cur.execute("DROP TABLE IF EXISTS kv")
        cur.execute("CREATE TABLE kv (meta JSONB)")
        # 写法一：用 Jsonb 包一个 Python 字典，最推荐
        cur.execute("INSERT INTO kv (meta) VALUES (%s)", (Jsonb({"a": 1, "b": [1, 2]}),))
        # 写法二：直接传 JSON 字符串，也可以
        cur.execute("INSERT INTO kv (meta) VALUES (%s)", ('{"a": 2}',))
        conn.commit()
        # 读出来自动变成 Python 字典，不用手动 json.loads
        cur.execute("SELECT meta FROM kv ORDER BY meta -> 'a'")
        for (meta,) in cur.fetchall():
            print(meta, type(meta).__name__)

输出：

{'a': 1, 'b': [1, 2]} dict
{'a': 2} dict

8.9. 完整脚本与输出 #

把这一节的类型用法串成一个完整脚本。存成 pg_types.py：

"""常用数据类型与 JSONB 实战。

可反复运行。演示：自增主键、时间类型、数组、枚举、JSONB 的存取改查。
"""

# 驱动
import psycopg

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

# 建立连接，with 结束时自动关闭
with psycopg.connect(DSN) as conn:
    # 拿一个游标
    with conn.cursor() as cur:
        # ---------- 1. 建表：把常用类型都用一遍 ----------
        # 先删旧表，让脚本可以反复运行
        cur.execute("DROP TABLE IF EXISTS articles")
        # 枚举类型要先删再建，否则第二次运行会报「类型已存在」
        cur.execute("DROP TYPE IF EXISTS article_status")
        # CREATE TYPE ... AS ENUM 定义一个只能取固定几个值的类型
        cur.execute("CREATE TYPE article_status AS ENUM ('draft', 'published', 'archived')")
        # 建表，每个字段演示一种常用类型
        cur.execute("""
            CREATE TABLE articles (
                -- 自增主键，数据库自动填值
                id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                -- 变长字符串，PostgreSQL 里首选 TEXT 而不是 VARCHAR(n)
                title       TEXT           NOT NULL,
                -- 整数，不填默认 0
                word_count  INT            NOT NULL DEFAULT 0,
                -- 精确小数：总共 4 位数字，其中 2 位小数，也就是最大 99.99
                score       NUMERIC(4, 2),
                -- 布尔值，只能是 true / false
                is_top      BOOLEAN        NOT NULL DEFAULT false,
                -- 枚举类型，只能取上面 CREATE TYPE 里定义的那三个值
                status      article_status NOT NULL DEFAULT 'draft',
                -- 字符串数组，方括号是数组的标志
                tags        TEXT[],
                -- JSONB：结构不固定的数据放这里，默认给个空对象避免 NULL
                meta        JSONB          NOT NULL DEFAULT '{}'::jsonb,
                -- 带时区的时间戳，可以为空表示还没发布
                published_at TIMESTAMPTZ
            )
        """)
        # 提交建表
        conn.commit()
        print("articles 表已建好")

        # ---------- 2. 插入：Python 类型如何对应 PostgreSQL 类型 ----------
        cur.execute(
            """
            -- 列出要写入的八个字段
            INSERT INTO articles (title, word_count, score, is_top, status, tags, meta, published_at)
            -- 前七个用占位符从 Python 传，最后一个直接用数据库的 now()
            VALUES (%s, %s, %s, %s, %s, %s, %s, now())
            -- 顺便把生成的 id 返回
            RETURNING id
            """,
            (
                # TEXT 收 Python 的 str
                "PostgreSQL 入门",
                # INT 收 Python 的 int
                3200,
                # NUMERIC 收 float 或 Decimal，这里用 float
                4.75,
                # BOOLEAN 收 Python 的 bool
                True,
                # 枚举字段收字符串，值必须是定义好的那几个之一
                "published",
                # TEXT[] 数组字段直接收 Python 列表
                ["数据库", "入门"],
                # JSONB 字段可以直接收 Python 字典，psycopg 会自动转成 JSON
                psycopg.types.json.Jsonb({"author": "张三", "level": 1, "topics": ["SQL", "索引"]}),
            ),
        )
        print("插入的 id:", cur.fetchone()[0])
        conn.commit()

        # ---------- 3. 读出来是什么 Python 类型 ----------
        cur.execute("SELECT title, score, is_top, status, tags, meta, published_at FROM articles")
        # 取出唯一那一行
        row = cur.fetchone()
        # 逐个打印值和它对应的 Python 类型
        for name, value in zip(
            ["title", "score", "is_top", "status", "tags", "meta", "published_at"], row
        ):
            print(f"  {name:13s} = {str(value)[:46]:48s} ({type(value).__name__})")

        # ---------- 4. JSONB：取值 ----------
        print("\n-- JSONB 取值 --")
        # ->> 取出来是文本，-> 取出来还是 JSON
        cur.execute("SELECT meta ->> 'author' AS 文本, meta -> 'topics' AS json值 FROM articles")
        print("  ->> 与 ->:", cur.fetchone())
        # #>> 按路径取深层的值，路径写成花括号数组
        cur.execute("SELECT meta #>> '{topics,0}' FROM articles")
        print("  按路径取第一个 topic:", cur.fetchone()[0])

        # ---------- 5. JSONB：条件查询 ----------
        print("\n-- JSONB 条件查询 --")
        # @> 判断「左边是否包含右边这个片段」，最常用
        cur.execute("SELECT title FROM articles WHERE meta @> %s", ('{"author": "张三"}',))
        print("  @> 包含:", cur.fetchall())
        # ? 判断某个键是否存在
        cur.execute("SELECT title FROM articles WHERE meta ? 'level'")
        print("  ? 键存在:", cur.fetchall())
        # JSON 里的值都是 JSON 类型，比大小要先 ::int 转成数字
        cur.execute("SELECT title FROM articles WHERE (meta ->> 'level')::int >= 1")
        print("  数字比较（先转类型）:", cur.fetchall())
        # 对 JSON 数组用 ? 判断是否含某个元素
        cur.execute("SELECT title FROM articles WHERE meta -> 'topics' ? 'SQL'")
        print("  数组含某元素:", cur.fetchall())

        # ---------- 6. JSONB：修改 ----------
        print("\n-- JSONB 修改 --")
        # jsonb_set 改指定路径的值，第三个参数必须是合法 JSON 文本
        cur.execute("UPDATE articles SET meta = jsonb_set(meta, '{level}', '3') RETURNING meta")
        print("  jsonb_set 改 level:", cur.fetchone()[0])
        # || 合并两个 JSONB，右边的键会覆盖左边同名键，可用来「加字段」
        cur.execute("""UPDATE articles SET meta = meta || '{"lang": "zh"}'::jsonb RETURNING meta""")
        print("  || 加字段:", cur.fetchone()[0])
        # - 删掉一个键
        cur.execute("UPDATE articles SET meta = meta - 'lang' RETURNING meta")
        print("  - 删字段:", cur.fetchone()[0])
        conn.commit()

        # ---------- 7. 数组字段的操作 ----------
        print("\n-- 数组字段 --")
        # @> 对数组同样是「包含」
        cur.execute("SELECT title FROM articles WHERE tags @> ARRAY['入门']")
        print("  数组包含:", cur.fetchall())
        # array_length 取长度，第二个参数 1 表示第一维
        cur.execute("SELECT array_length(tags, 1) FROM articles")
        print("  数组长度:", cur.fetchone()[0])
        # unnest 把数组展开成多行，便于统计
        cur.execute("SELECT unnest(tags) AS tag FROM articles ORDER BY tag")
        print("  展开成行:", cur.fetchall())

实测输出：

articles 表已建好
插入的 id: 1
  title         = PostgreSQL 入门                                    (str)
  score         = 4.75                                             (Decimal)
  is_top        = True                                             (bool)
  status        = published                                        (str)
  tags          = ['数据库', '入门']                                    (list)
  meta          = {'level': 1, 'author': '张三', 'topics': ['SQL',   (dict)
  published_at  = 2026-08-15 01:17:30.580291+08:00                 (datetime)

-- JSONB 取值 --
  ->> 与 ->: ('张三', ['SQL', '索引'])
  按路径取第一个 topic: SQL

-- JSONB 条件查询 --
  @> 包含: [('PostgreSQL 入门',)]
  ? 键存在: [('PostgreSQL 入门',)]
  数字比较（先转类型）: [('PostgreSQL 入门',)]
  数组含某元素: [('PostgreSQL 入门',)]

-- JSONB 修改 --
  jsonb_set 改 level: {'level': 3, 'author': '张三', 'topics': ['SQL', '索引']}
  || 加字段: {'lang': 'zh', 'level': 3, 'author': '张三', 'topics': ['SQL', '索引']}
  - 删字段: {'level': 3, 'author': '张三', 'topics': ['SQL', '索引']}

-- 数组字段 --
  数组包含: [('PostgreSQL 入门',)]
  数组长度: 2
  展开成行: [('入门',), ('数据库',)]

两个细节值得注意。 一是 score 传进去是 4.75 这个 float，读出来变成了 Decimal——因为列类型是 NUMERIC，读出来的类型由列类型决定，不由你传进去的类型决定。二是 meta 打印出来键的顺序变成了 level, author, topics，和插入时的 author, level, topics 不一样，这就是 8.5 节说的「JSONB 不保留键顺序」。

9. 索引 #

第 3.6 节用「书末尾的索引页」类比过索引。这一节用 10 万行真实数据把效果测出来，让你知道该在什么时候加、加完怎么验证。

9.1. 没有索引时发生了什么 #

先准备一张 10 万行的表，user_id 列上故意不建索引，然后查其中一个值：

-- EXPLAIN 让 PostgreSQL 说出「它打算怎么执行这条查询」
-- 加上 ANALYZE 会真的执行一遍，然后给出真实耗时
EXPLAIN (ANALYZE) SELECT * FROM events WHERE user_id = 12345;

实测输出：

Seq Scan on events  (cost=0.00..1874.00 rows=5 width=17) (actual time=0.060..4.341 rows=9.00 loops=1)
  Filter: (user_id = 12345)
  Rows Removed by Filter: 99991
  Buffers: shared hit=624
Planning Time: 1.131 ms
Execution Time: 4.360 ms

这段输出里有三个关键信息，学会看它们比记住任何优化技巧都有用：

Seq Scan on events 意思是「顺序扫描 events 表」，也就是从第一行翻到最后一行。这是没有索引时唯一的选择。

Rows Removed by Filter: 99991 是最能说明问题的一行：为了找到那 9 行，它检查了 10 万行，然后把不符合条件的 99991 行扔掉了。白干了 99.99% 的活。

Execution Time: 4.360 ms 是真实耗时。10 万行 4 毫秒看着不慢，但注意这个时间跟数据量成正比：涨到 1000 万行就是 400 毫秒，再加上并发就撑不住了。

9.2. 建索引之后 #

在 user_id 上建一个索引：

-- CREATE INDEX 索引名 ON 表名 (列名)
-- 索引名习惯写成 idx_表名_列名，方便日后辨认
CREATE INDEX idx_events_user_id ON events (user_id);

-- 建完索引后建议跑一次 ANALYZE，让优化器更新这张表的统计信息
-- 不跑的话它可能因为统计过时而选错执行方案
ANALYZE events;

跑完全一样的查询：

-- 同一条查询，一个字都没改
EXPLAIN (ANALYZE) SELECT * FROM events WHERE user_id = 12345;

实测输出：

Bitmap Heap Scan on events  (cost=4.33..23.05 rows=5 width=17) (actual time=0.050..0.058 rows=9.00 loops=1)
  Recheck Cond: (user_id = 12345)
  Heap Blocks: exact=9
  ->  Bitmap Index Scan on idx_events_user_id  (cost=0.00..4.33 rows=5 width=0) (actual time=0.036..0.036 rows=9.00 loops=1)
        Index Cond: (user_id = 12345)
        Index Searches: 1
Planning Time: 0.147 ms
Execution Time: 0.072 ms

对比一下这两次的差别：

	没有索引	有索引
扫描方式	Seq Scan（全表扫）	Bitmap Index Scan（走索引）
白扫的行数	99991 行	0 行
执行耗时	4.360 ms	0.072 ms

快了约 60 倍，而且这个倍数会随数据量继续放大——因为全表扫描的耗时跟行数成正比，而 B-tree 索引的查找次数只跟行数的对数成正比，也就是 $O(\log n)$。数据翻 10 倍，全表扫描慢 10 倍，索引查找只慢一点点。

9.3. 怎么读 EXPLAIN 的输出 #

EXPLAIN 的输出看着复杂，但刚开始只要认准几个词就够用了。先看扫描方式：

看到这个	意思	好还是坏
Seq Scan	全表逐行扫	小表没关系；大表上出现就要警惕
Index Scan	走索引，直接定位	好
Bitmap Index Scan	先用索引攒一批位置，再批量去取	好，命中多行时比 Index Scan 更优
Index Only Scan	只读索引就够了，连表都不用碰	最好

再看两个数字：

• Rows Removed by Filter —— 白扫了多少行。这个数字很大就说明有优化空间。
• Execution Time —— 真实耗时，优化前后就对比这个。

还有一个容易被误读的地方： 输出里的 rows=5 是优化器的估算，而 actual ... rows=9.00 是真实行数。上面的例子里估 5 实际 9，很接近，说明统计信息是准的。如果这两个数差了几十倍（比如估 5 实际 5000），通常意味着该跑 ANALYZE 更新统计信息了——估错会让优化器选到很差的执行方案。

9.4. 哪些写法会让索引失效 #

索引建了不等于用得上。 最常见的失效原因是「对列做了运算」——数据库看到的不再是那一列本身，而是一个表达式，列上的索引就派不上用场了：

-- 失效写法：user_id + 0 在数学上等于 user_id，但索引用不了
EXPLAIN (ANALYZE) SELECT * FROM events WHERE user_id + 0 = 12345;

实测输出，又变回全表扫描了：

Seq Scan on events  (cost=0.00..2124.00 rows=500 width=17) (actual time=0.059..4.768 rows=9.00 loops=1)
  Filter: ((user_id + 0) = 12345)
  Rows Removed by Filter: 99991
Execution Time: 4.774 ms

同一类的常见失效写法，以及怎么改：

失效写法	为什么	改成
WHERE age + 1 > 30	列上有运算	WHERE age > 29
WHERE upper(name) = 'ABC'	列上套了函数	建表达式索引，见下
WHERE name LIKE '%三'	开头就是通配符，没法用前缀定位	用全文检索或 pg_trgm 扩展
WHERE created_at::date = '2026-08-15'	列上有类型转换	WHERE created_at >= '2026-08-15' AND created_at < '2026-08-16'

如果确实需要按函数结果查，可以直接给表达式建索引：

-- 表达式索引：索引的是 upper(name) 的结果，而不是 name 本身
-- 这样 WHERE upper(name) = 'ABC' 就能用上索引了
CREATE INDEX idx_users_upper_name ON users (upper(name));

LIKE '前缀%' 这种「通配符在后面」的是可以用索引的，只有 '%后缀' 这种开头就通配的不行——道理和查字典一样，知道首字母才能定位。

9.5. 命中太多时索引会被放弃 #

这一点很反直觉：即使有索引，数据库也可能主动不用它。 查一个能命中绝大多数行的条件：

-- user_id > 100 会命中 10 万行里的 9 万 9 千多行
EXPLAIN (ANALYZE) SELECT count(*) FROM events WHERE user_id > 100;

实测输出：

Aggregate  (cost=2122.72..2122.73 rows=1 width=8) (actual time=9.685..9.685 rows=1.00 loops=1)
  ->  Seq Scan on events  (cost=0.00..1874.00 rows=99486 width=0) (actual time=0.008..6.232 rows=99509.00 loops=1)
        Filter: (user_id > 100)
        Rows Removed by Filter: 491
Execution Time: 9.695 ms

明明 user_id 上有索引，它却选了 Seq Scan。这是对的选择，不是 bug。 走索引的流程是「查索引拿到位置 → 回表取数据」，两步。当你要取的是表里 99.5% 的行时，这两步的总开销比直接顺序读一遍整张表更大——顺序读磁盘比东跳西跳地随机读快得多。

这条规律的实用含义是：索引只对「筛得很狠」的查询有效。 一个只有「男/女」两个值的性别列，建索引几乎没用，因为任何一个值都命中一半的数据。

9.6. JSONB 用 GIN 索引 #

B-tree 索引对 JSONB 的 @> 这类查询无效，要用 GIN 索引：

-- USING GIN 指定索引类型；不写 USING 就是默认的 B-tree
CREATE INDEX idx_big_docs_meta ON big_docs USING GIN (meta);

第 9.5 节讲的「选择性」规律在 GIN 上体现得更明显。下面这个脚本灌 5 万行 JSONB 数据，然后对比三种情况。存成 pg_gin.py，可反复运行：

"""JSONB 的 GIN 索引实测：为什么「选择性」决定索引值不值。

可反复运行。会自己灌 5 万行 JSONB 数据、建 GIN 索引、对比两种查询。
"""

# 生成随机数据
import random

# 驱动
import psycopg

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"
# 灌多少行
ROWS = 50_000


# 跑一次 EXPLAIN ANALYZE，只把关键两行摘出来
def brief(cur: psycopg.Cursor, sql: str) -> None:
    """打印执行计划里的扫描方式和真实耗时。"""
    # ANALYZE 会真的执行一遍，拿到真实数字
    cur.execute(f"EXPLAIN (ANALYZE) {sql}")
    # 每行计划是单元素元组，摊平成字符串
    lines = [r[0] for r in cur.fetchall()]
    # 找出扫描方式所在的那一行
    scan = next((x.strip() for x in lines if "Scan" in x), "?")
    # 找出真实耗时那一行
    took = next((x.strip() for x in lines if "Execution Time" in x), "?")
    # 打印出来
    print(f"      {scan}")
    print(f"      {took}")


# 建立连接和游标
with psycopg.connect(DSN) as conn, conn.cursor() as cur:
    # ---------- 1. 准备 5 万行 JSONB 数据 ----------
    # 反复运行时先删表
    cur.execute("DROP TABLE IF EXISTS big_docs")
    # meta 是 JSONB 列，里面放一个部门和一个流水号
    cur.execute("""
        CREATE TABLE big_docs (
            id   BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
            meta JSONB NOT NULL
        )
    """)
    conn.commit()

    # 固定随机种子，保证每次运行数据一致
    random.seed(7)
    # 只有五个部门，所以 dept 这个键的「选择性」很低
    depts = ["hr", "finance", "it", "sales", "legal"]
    # 用 COPY 快速灌数据
    with cur.copy("COPY big_docs (meta) FROM STDIN") as cp:
        # 每行造一个 JSON 字符串；no 是流水号，选择性很高
        for i in range(ROWS):
            cp.write_row([f'{{"dept": "{random.choice(depts)}", "no": {i}}}'])
    conn.commit()
    # 更新统计信息，让优化器做出正确判断
    cur.execute("ANALYZE big_docs")
    conn.commit()
    print(f"已灌入 {ROWS:,} 行 JSONB 数据")

    # ---------- 2. 没有索引时 ----------
    print("\n【没有 GIN 索引】按 dept 查")
    brief(cur, """SELECT count(*) FROM big_docs WHERE meta @> '{"dept": "legal"}'""")

    # ---------- 3. 建 GIN 索引 ----------
    # USING GIN 指定索引类型；JSONB 的 @> 查询必须用 GIN，B-tree 用不上
    cur.execute("CREATE INDEX idx_big_docs_meta ON big_docs USING GIN (meta)")
    conn.commit()
    # 建完再统计一次
    cur.execute("ANALYZE big_docs")
    conn.commit()

    # ---------- 4. 有索引，但命中很多行 ----------
    print("\n【有 GIN 索引】按 dept 查：只有 5 个部门，命中约 1 万行")
    brief(cur, """SELECT count(*) FROM big_docs WHERE meta @> '{"dept": "legal"}'""")

    # ---------- 5. 有索引，且命中很少行 ----------
    print("\n【有 GIN 索引】按 no 查：流水号唯一，只命中 1 行")
    brief(cur, """SELECT count(*) FROM big_docs WHERE meta @> '{"no": 42}'""")

    # ---------- 6. 看索引占了多少空间 ----------
    print("\n【空间代价】GIN 索引比 B-tree 大不少")
    # pg_size_pretty 把字节数转成人能读的单位
    cur.execute("""
        SELECT pg_size_pretty(pg_relation_size('big_docs'))          AS 表,
               pg_size_pretty(pg_relation_size('idx_big_docs_meta')) AS 索引
    """)
    size = cur.fetchone()
    print(f"      表 {size[0]}，GIN 索引 {size[1]}")

实测输出：

已灌入 50,000 行 JSONB 数据

【没有 GIN 索引】按 dept 查
      ->  Seq Scan on big_docs  (cost=0.00..1153.00 rows=7576 width=0) (actual time=0.016..7.259 rows=10002.00 loops=1)
      Execution Time: 7.645 ms

【有 GIN 索引】按 dept 查：只有 5 个部门，命中约 1 万行
      ->  Bitmap Heap Scan on big_docs  (cost=78.99..714.32 rows=8586 width=0) (actual time=1.001..3.570 rows=10002.00 loops=1)
      Execution Time: 3.906 ms

【有 GIN 索引】按 no 查：流水号唯一，只命中 1 行
      ->  Bitmap Heap Scan on big_docs  (cost=21.57..40.17 rows=5 width=0) (actual time=0.034..0.034 rows=1.00 loops=1)
      Execution Time: 0.045 ms

【空间代价】GIN 索引比 B-tree 大不少
      表 4224 kB，GIN 索引 2984 kB

这组数字有三个结论：

第一，按 dept 查，GIN 索引只带来了不到 2 倍的提升（7.645 ms → 3.906 ms）。因为只有 5 个部门，任何一个部门都命中五分之一的数据，索引省不了多少事。

第二，同一个索引，按 no 查快了 87 倍（3.906 ms → 0.045 ms）。流水号是唯一的，一查就定位到 1 行。索引没变，只是查询的选择性变了。

第三，GIN 索引占了 2984 kB，是表本身（4224 kB）的 71%。对比第 9.7 节 B-tree 索引只占 23%，GIN 的空间代价明显更高。

所以 JSONB 里该建索引的是「值很分散」的键（订单号、用户 id、文档 id），而不是「只有几种取值」的键（状态、类型、部门）。

如果只需要给 JSONB 里某一个键加速，比给整个 meta 建 GIN 更省空间的办法是给那个键建 B-tree 表达式索引：

-- 只给 meta 里的 dept 键建索引，比给整个 meta 建 GIN 小得多
-- 注意 ->> 两边要用括号包起来，否则语法报错
CREATE INDEX idx_big_docs_dept ON big_docs ((meta ->> 'dept'));

9.7. 索引的代价 #

索引不是免费的。看一下刚才那个索引占了多少空间：

-- pg_relation_size 返回字节数，pg_size_pretty 把它变成人能读的单位
SELECT pg_size_pretty(pg_relation_size('events'))             AS 表大小,
       pg_size_pretty(pg_relation_size('idx_events_user_id')) AS 索引大小;

实测输出：

 表大小  | 索引大小
---------+----------
 4992 kB | 1168 kB

一个单列索引就占了表本身 23% 的空间。 而且代价不止空间：

1. 写入变慢。 每次 INSERT / UPDATE / DELETE 都要顺带维护所有相关索引。一张表上挂十个索引，写入速度会有明显下降。
2. 占内存。 索引也要缓存在内存里才快，索引太多会挤占宝贵的缓存空间。

所以索引的正确用法是「按需建」，不是「都建上」。给出三条判断标准：

该建	不该建
经常出现在 WHERE 里的列	很少用来过滤的列
值很分散的列（id、邮箱、订单号）	只有几种取值的列（性别、状态）
经常用来 ORDER BY 或做表连接的列	表本身只有几百行（全表扫更快）

最后一个提醒：主键和 UNIQUE 约束会自动带索引，不用手动再建（第 5.2 节 \d users 的输出里能看到）。给主键重复建索引是常见的浪费。

9.8. 完整脚本与输出 #

存成 pg_index.py，直接运行就能复现上面所有实测：

"""索引效果实测：10 万行数据下，有索引和没索引差多少。

可反复运行。会自己灌数据、自己建索引、自己打印执行计划。
"""

# 用来生成随机测试数据
import random
# 用来计时
import time

# 驱动
import psycopg

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"
# 灌多少行测试数据；十万行足够让全表扫描和索引扫描分出胜负
ROWS = 100_000


def explain(cur: psycopg.Cursor, sql: str) -> tuple[str, float]:
    """跑一次 EXPLAIN ANALYZE，返回（用了什么扫描方式，耗时毫秒）。"""
    # EXPLAIN ANALYZE 会真的执行语句，然后把执行计划和真实耗时一起返回
    cur.execute(f"EXPLAIN (ANALYZE) {sql}")
    # 每一行计划是一个单元素元组，先摊平成字符串列表
    lines = [r[0] for r in cur.fetchall()]
    # 找出第一行里的扫描方式，它决定了快慢
    scan = next((x.strip().split("  ")[0] for x in lines if "Scan" in x), "?")
    # 最后的 Execution Time 是真实执行耗时
    ms = float(next(x for x in lines if "Execution Time" in x).split()[-2])
    # 顺便把「白扫了多少行」也找出来，这个数字最能说明问题
    removed = next((x.strip() for x in lines if "Rows Removed by Filter" in x), "")
    # 打印完整计划，方便对照学习
    for line in lines:
        print("      " + line)
    if removed:
        print(f"      （注意上面这行：{removed}）")
    return scan, ms


# 建立连接
with psycopg.connect(DSN) as conn:
    # 拿游标
    with conn.cursor() as cur:
        # ---------- 1. 准备一张有点体量的表 ----------
        # 反复运行时先删掉旧表
        cur.execute("DROP TABLE IF EXISTS events")
        # 建表：user_id 是我们后面要查的列，先故意不建索引
        cur.execute("""
            CREATE TABLE events (
                id      BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                user_id INT  NOT NULL,
                action  TEXT NOT NULL
            )
        """)
        conn.commit()

        # 固定随机种子，保证每次运行的数据分布一致
        random.seed(42)
        # 先在内存里造好所有行
        rows = [(random.randint(1, 20_000), random.choice(["login", "click", "buy"])) for _ in range(ROWS)]
        # 记下开始时间
        t0 = time.perf_counter()
        # COPY 是 PostgreSQL 的批量导入通道，比逐条 INSERT 快一两个数量级
        with cur.copy("COPY events (user_id, action) FROM STDIN") as cp:
            # 逐行写进 COPY 流
            for r in rows:
                cp.write_row(r)
        conn.commit()
        print(f"COPY 灌入 {ROWS:,} 行，耗时 {time.perf_counter() - t0:.2f} 秒")

        # ANALYZE 让优化器重新统计这张表的数据分布，不做的话它可能选错方案
        cur.execute("ANALYZE events")
        conn.commit()

        # ---------- 2. 没有索引：全表扫描 ----------
        print("\n【没有索引】查 user_id = 12345")
        scan1, ms1 = explain(cur, "SELECT * FROM events WHERE user_id = 12345")

        # ---------- 3. 建索引 ----------
        # 在 user_id 上建默认的 B-tree 索引
        cur.execute("CREATE INDEX idx_events_user_id ON events (user_id)")
        conn.commit()
        # 建完索引再统计一次
        cur.execute("ANALYZE events")
        conn.commit()

        # ---------- 4. 有索引：索引扫描 ----------
        print("\n【建了索引】同一条查询")
        scan2, ms2 = explain(cur, "SELECT * FROM events WHERE user_id = 12345")

        # ---------- 5. 对比 ----------
        print(f"\n对比：{scan1} {ms1} ms  ->  {scan2} {ms2} ms，快了约 {ms1 / ms2:.1f} 倍")

        # ---------- 6. 索引失效的典型写法 ----------
        print("\n【索引失效】对列做了运算，索引就用不上了")
        # user_id + 0 在数学上等于 user_id，但数据库看到的是「一个表达式」，没法用列上的索引
        explain(cur, "SELECT * FROM events WHERE user_id + 0 = 12345")

        # ---------- 7. 范围查询也能用索引 ----------
        print("\n【范围查询】B-tree 索引对 BETWEEN 同样有效")
        explain(cur, "SELECT count(*) FROM events WHERE user_id BETWEEN 100 AND 200")

        # ---------- 8. 查得太多时，索引反而不划算 ----------
        print("\n【命中太多】要返回大半张表时，优化器会主动放弃索引")
        explain(cur, "SELECT count(*) FROM events WHERE user_id > 100")

        # ---------- 9. 看看索引占了多少空间 ----------
        print("\n【空间代价】索引不是免费的")
        # pg_size_pretty 把字节数变成人能读的单位
        cur.execute("""
            SELECT pg_size_pretty(pg_relation_size('events'))       AS 表,
                   pg_size_pretty(pg_relation_size('idx_events_user_id')) AS 索引
        """)
        size = cur.fetchone()
        print(f"      表 {size[0]}，索引 {size[1]}")

实测输出（为节省篇幅，只保留关键行）：

COPY 灌入 100,000 行，耗时 0.24 秒

【没有索引】查 user_id = 12345
      Seq Scan on events  (cost=0.00..1874.00 rows=5 width=17) (actual time=0.060..4.341 rows=9.00 loops=1)
        Filter: (user_id = 12345)
        Rows Removed by Filter: 99991
      Execution Time: 4.360 ms
      （注意上面这行：Rows Removed by Filter: 99991）

【建了索引】同一条查询
      Bitmap Heap Scan on events  (cost=4.33..23.05 rows=5 width=17) (actual time=0.050..0.058 rows=9.00 loops=1)
        ->  Bitmap Index Scan on idx_events_user_id  (cost=0.00..4.33 rows=5 width=0)
              Index Cond: (user_id = 12345)
      Execution Time: 0.072 ms

对比：Seq Scan on events 4.36 ms  ->  Bitmap Heap Scan on events 0.072 ms，快了约 60.6 倍

【索引失效】对列做了运算，索引就用不上了
      Seq Scan on events  (cost=0.00..2124.00 rows=500 width=17) (actual time=0.059..4.768 rows=9.00 loops=1)
        Filter: ((user_id + 0) = 12345)
        Rows Removed by Filter: 99991
      Execution Time: 4.774 ms

【范围查询】B-tree 索引对 BETWEEN 同样有效
      ->  Bitmap Heap Scan on events  (cost=9.61..641.04 rows=519 width=0) (actual time=0.091..0.282 rows=527.00 loops=1)
              Index Cond: ((user_id >= 100) AND (user_id <= 200))
      Execution Time: 0.336 ms

【命中太多】要返回大半张表时，优化器会主动放弃索引
      ->  Seq Scan on events  (cost=0.00..1874.00 rows=99486 width=0) (actual time=0.008..6.232 rows=99509.00 loops=1)
              Filter: (user_id > 100)
              Rows Removed by Filter: 491
      Execution Time: 9.695 ms

【空间代价】索引不是免费的
      表 4992 kB，索引 1168 kB

10. 连接池 #

前面九节讲的都是「怎么把 SQL 写对」。这一节讲的是一件跟 SQL 完全无关、但对性能影响可能更大的事：怎么管理连接。

这一节值得认真看，原因是它的收益比优化 SQL 更容易拿到。第 9 节费力建索引换来的是零点几毫秒的查询提升，而这一节改几行代码就能省掉每次请求 50 毫秒的固定开销。对 PostgreSQL 来说，连接池不是「进阶优化」，而是基本配置。

10.1. 为什么 PostgreSQL 特别需要它 #

PostgreSQL 有一个设计特点：每来一个客户端连接，服务端就开一个独立的操作系统进程去伺候它。

这个设计让它非常稳（一个连接崩了不影响别人），但代价是建立连接很贵——不是简单地开个网络端口，而是要创建进程、分配内存、初始化一堆状态。

后果有两个：

1. 每次新建连接都要等几十毫秒。 如果你的接口每次请求都新建一个连接，光这一步就把响应时间吃掉一大截。
2. 连接数有硬上限。 默认 max_connections = 100，超了就直接连不上。而每个连接还要占几 MB 内存，也不是简单调大就行。

连接池就是解决这两个问题的标准做法： 程序启动时先建好几个连接放在池子里，用的时候借一个、用完还回去，而不是每次都新建再销毁。

10.2. 实测：256 倍 #

这个差距有多大值得亲眼看一次。同样做 30 次极简单的查询，一组每次新建连接，一组从池里借：

# 计时
import time

# 驱动
import psycopg
# 连接池在单独的包里：pip install "psycopg[pool]"
from psycopg_pool import ConnectionPool

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"
# 各做多少次查询
N = 30

# ---------- 第一组：每次都新建连接 ----------
t0 = time.perf_counter()
for _ in range(N):
    # 每一轮都要走完整的握手：TCP 连接、身份验证、服务端开一个新进程
    with psycopg.connect(DSN) as conn:
        with conn.cursor() as cur:
            # 查询本身极简单，所以测出来的基本就是连接开销
            cur.execute("SELECT 1")
            cur.fetchone()
no_pool = time.perf_counter() - t0
print(f"每次新建连接，{N} 次共 {no_pool:.3f} 秒")

# ---------- 第二组：用连接池 ----------
# 建池：min_size 是常备连接数，open=True 表示建池时就把连接准备好
pool = ConnectionPool(DSN, min_size=2, max_size=5, open=True)
# 等常备连接就绪，避免把建池时间算进测试
pool.wait()
t0 = time.perf_counter()
for _ in range(N):
    # pool.connection() 是「借一个」，with 结束时还回池里而不是关闭
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT 1")
            cur.fetchone()
pooled = time.perf_counter() - t0
print(f"连接池复用，  {N} 次共 {pooled:.3f} 秒")
print(f"差距：{no_pool / pooled:.0f} 倍")
# 用完关池，否则后台线程不退出
pool.close()

实测输出：

每次新建连接，30 次共 1.546 秒
连接池复用，  30 次共 0.006 秒
差距：256 倍

换算成单次：新建连接平均 51.5 毫秒，从池里借平均 0.20 毫秒。

这个数字的实际意义是：如果你的接口每次都新建连接，那么在任何 SQL 执行之前，就已经先花掉了 50 毫秒。 而一个索引命中的查询本身只要 0.07 毫秒（第 9.2 节实测）——连接开销是查询开销的七百倍。优化 SQL 之前，先确认自己用了连接池，性价比高得多。

10.3. 池子开多大 #

两个参数要设：

参数	含义	怎么设
min_size	常备连接数，池子空闲时也保持这么多	2～5
max_size	最大连接数，忙的时候最多开到这么多	见下面的公式

max_size 的设定要从服务端的上限倒推，而不是随手写个大数字：

$$ \text{max_size} \times \text{进程数} \le \text{max_connections} - \text{留给运维的余量} $$

举个例子：服务端 max_connections = 100，留 10 个给你自己用 psql 排查问题，你的服务部署了 4 个进程，那么每个进程的 max_size 最多设到 $(100 - 10) / 4 \approx 22$。

初学者最容易犯的错是把 max_size 设得过大，觉得「越大越能扛」。实际上连接数超过 CPU 核数太多之后，吞吐量不升反降——因为进程都在抢 CPU 和内存。对多数应用，max_size 设成 10～20 就够了。

看服务端当前的连接情况：

-- pg_stat_activity 是服务端所有连接的实时视图
-- 排查「连接数爆了」的问题时先看这个
SELECT count(*) AS 当前连接数 FROM pg_stat_activity;

-- 看上限是多少
SHOW max_connections;

10.4. 完整脚本与输出 #

存成 pg_pool.py：

"""连接池实测：为什么用 PostgreSQL 基本一定要配连接池。

可反复运行，不改任何数据，只做连接开销对比。
"""

# 计时
import time

# 驱动
import psycopg
# 连接池在单独的包里：pip install "psycopg[pool]"
from psycopg_pool import ConnectionPool

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"
# 各做多少次「查一下」，次数太少看不出差别
N = 30


def without_pool() -> float:
    """每次查询都新建一个连接，用完就关。"""
    # 计时开始
    t0 = time.perf_counter()
    # 循环 N 次
    for _ in range(N):
        # 每一轮都要走完整的握手：TCP 连接、身份验证、服务端 fork 一个新进程
        with psycopg.connect(DSN) as conn:
            with conn.cursor() as cur:
                # 查询本身极简单，耗时几乎可以忽略，所以测出来的就是连接开销
                cur.execute("SELECT 1")
                cur.fetchone()
    # 返回总耗时
    return time.perf_counter() - t0


def with_pool() -> tuple[float, dict]:
    """从连接池里借连接，用完还回去。"""
    # 建池：min_size 是常备连接数，max_size 是上限
    # open=True 表示创建时就把常备连接建好
    pool = ConnectionPool(DSN, min_size=2, max_size=5, open=True)
    # 等常备连接都就绪，避免把建池时间算进测试
    pool.wait()
    # 计时开始
    t0 = time.perf_counter()
    # 同样循环 N 次
    for _ in range(N):
        # pool.connection() 是「借一个」，with 结束时自动还回池里而不是关闭
        with pool.connection() as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT 1")
                cur.fetchone()
    # 记下耗时
    elapsed = time.perf_counter() - t0
    # 取一下池的统计信息
    stats = pool.get_stats()
    # 用完记得关池，否则后台线程不退出
    pool.close()
    return elapsed, stats


# 直接运行时才执行
if __name__ == "__main__":
    # 先测不用池
    t_no = without_pool()
    print(f"每次新建连接，{N} 次共 {t_no:.3f} 秒，平均 {t_no / N * 1000:.1f} 毫秒/次")

    # 再测用池
    t_yes, stats = with_pool()
    print(f"连接池复用，  {N} 次共 {t_yes:.3f} 秒，平均 {t_yes / N * 1000:.2f} 毫秒/次")

    # 算个倍数，这个数字很能说明问题
    print(f"差距：{t_no / t_yes:.0f} 倍")
    # 池里实际建了几个连接
    print(f"池里的连接数：{stats.get('pool_size')}（复用 {N} 次只用了这么多）")

    # 顺便看看服务端的连接情况
    with psycopg.connect(DSN) as conn:
        with conn.cursor() as cur:
            # pg_stat_activity 是服务端当前所有连接的视图
            cur.execute("SELECT count(*) FROM pg_stat_activity")
            used = cur.fetchone()[0]
            # max_connections 是服务端允许的连接上限，默认 100
            cur.execute("SHOW max_connections")
            limit = cur.fetchone()[0]
            print(f"服务端当前连接 {used} 个，上限 {limit} 个")

实测输出：

每次新建连接，30 次共 1.546 秒，平均 51.5 毫秒/次
连接池复用，  30 次共 0.006 秒，平均 0.20 毫秒/次
差距：256 倍
池里的连接数：2（复用 30 次只用了这么多）
服务端当前连接 9 个，上限 100 个

注意最后两行。 池里只建了 2 个连接（min_size 就是 2），却完成了 30 次查询——这就是复用的意思。而服务端一共才 9 个连接，离 100 的上限还很远。

11. 在 LangGraph 里用 PostgreSQL #

这一节是需要 PostgreSQL 的直接原因。

11.1. 两个位置：checkpointer 与 store #

LangGraph 里有两个地方要存数据，职责完全不同：

	checkpointer	store
存什么	每一步执行完的状态快照	你主动写进去的长期数据
按什么隔离	thread_id（一次会话）	命名空间（通常按用户）
生命周期	跟着这次会话	跨会话、跨 thread 长期保留
典型内容	对话历史、执行到哪一步了	用户姓名、城市、偏好
内存版	InMemorySaver	InMemoryStore
PostgreSQL 版	PostgresSaver	PostgresStore

一句话区分：checkpointer 管「这次聊到哪了」，store 管「关于这个用户我知道什么」。

换成 PostgreSQL 版之后，两者的数据都落到磁盘上，程序重启、换台机器跑，数据都还在——这是内存版做不到的。

11.2. PostgresSaver：存会话进度 #

用法只比内存版多一个 setup()：

# 给状态字段挂 reducer 用
import operator
# 类型标注
from typing import Annotated, TypedDict

# PostgreSQL 版的检查点存储
from langgraph.checkpoint.postgres import PostgresSaver
# 搭图的三件套
from langgraph.graph import END, START, StateGraph

# 连接串；sslmode=disable 是本机连接的常用写法，避免本地没配证书时报警
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo?sslmode=disable"


# 图的状态定义
class State(TypedDict):
    # 普通字段，后写的覆盖先写的
    n: int
    # 挂上 operator.add，新值会追加到旧列表后面而不是覆盖
    log: Annotated[list[str], operator.add]


def step1(state: State) -> dict:
    """第一步：加一。"""
    # 返回「部分更新」，只写要改的字段
    return {"n": state["n"] + 1, "log": ["step1"]}


def step2(state: State) -> dict:
    """第二步：乘十。"""
    return {"n": state["n"] * 10, "log": ["step2"]}


# from_conn_string 是上下文管理器，退出时自动关连接
with PostgresSaver.from_conn_string(DSN) as saver:
    # 第一次用必须调 setup()，它会建好需要的表；已经建过再调也不报错
    saver.setup()

    # 搭一个最简单的两步图
    g = StateGraph(State)
    g.add_node("step1", step1)
    g.add_node("step2", step2)
    g.add_edge(START, "step1")
    g.add_edge("step1", "step2")
    g.add_edge("step2", END)
    # 编译时把 saver 交给 checkpointer 参数，图就有了持久化能力
    app = g.compile(checkpointer=saver)

    # thread_id 标识「这是哪一次会话」，同一个 id 共享历史
    cfg = {"configurable": {"thread_id": "demo-thread-1"}}
    # 跑一次
    print("第一次跑:", app.invoke({"n": 1, "log": []}, cfg))
    # get_state 把存下来的状态从数据库读回来
    print("从库里读回:", app.get_state(cfg).values)

输出：

第一次跑: {'n': 20, 'log': ['step1', 'step2']}
从库里读回: {'n': 20, 'log': ['step1', 'step2']}

setup() 这一步不能省。 它负责建表，第一次连一个新库时必须调用。它是幂等的——已经建过再调不会报错也不会重复建，所以放在程序启动流程里是安全的。

11.3. PostgresStore：存长期记忆 #

store 的接口是「命名空间 + key」的键值存储，比 SQL 简单得多：

# PostgreSQL 版的长期记忆存储
from langgraph.store.postgres import PostgresStore

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo?sslmode=disable"

# 同样是上下文管理器
with PostgresStore.from_conn_string(DSN) as store:
    # 第一次用要先建表
    store.setup()

    # put(命名空间, key, 值)
    # 命名空间是元组，习惯上第一段放用户 id，第二段放数据种类
    store.put(("u001", "profile"), "main", {"name": "张三", "city": "杭州"})
    # 同一个用户下可以有多个命名空间
    store.put(("u001", "memories"), "note-1", {"text": "喜欢简洁回答"})
    # 换个用户写一条
    store.put(("u002", "profile"), "main", {"name": "李四", "city": "北京"})

    # get 按 key 精确取一条，返回的对象用 .value 拿内容
    print("u001 的档案:", store.get(("u001", "profile"), "main").value)
    # search 按命名空间列出多条
    print("u001 的笔记:", [(i.key, i.value) for i in store.search(("u001", "memories"))])
    # 不同命名空间天然隔离
    print("u002 的档案:", store.get(("u002", "profile"), "main").value)
    # list_namespaces 看现在有哪些命名空间
    print("所有命名空间:", store.list_namespaces())

输出：

u001 的档案: {'city': '杭州', 'name': '张三'}
u001 的笔记: [('note-1', {'text': '喜欢简洁回答'})]
u002 的档案: {'city': '北京', 'name': '李四'}
所有命名空间: [('u001', 'memories'), ('u001', 'profile'), ('u002', 'profile')]

命名空间的第一段放用户 id，是一个值得遵守的约定。 这样「查这个用户的所有数据」「删掉这个用户的所有数据」都是一句话的事，用户之间的隔离也是天然的——u001 的代码不可能不小心读到 u002 的数据。

11.4. 它们建了哪些表 #

两个 setup() 一共建了 6 张表。知道它们的存在，排查问题和清理数据时会方便很多：

-- information_schema.tables 是标准 SQL 的元数据视图
-- 这里只筛 LangGraph 建的那几张
SELECT table_name FROM information_schema.tables
WHERE table_schema = 'public'
  AND (table_name LIKE 'checkpoint%' OR table_name LIKE 'store%')
ORDER BY table_name;

实测输出：

 checkpoint_blobs
 checkpoint_migrations
 checkpoint_writes
 checkpoints
 store
 store_migrations

表	谁建的	存什么
checkpoints	PostgresSaver	状态快照的主表
checkpoint_blobs	PostgresSaver	大字段单独存
checkpoint_writes	PostgresSaver	每一步的写入记录
checkpoint_migrations	PostgresSaver	表结构版本，setup() 靠它判断要不要升级
store	PostgresStore	长期记忆的数据
store_migrations	PostgresStore	同上，版本管理

注意 store 表里命名空间的存法：元组 ("u001", "memories") 被拼成了点号分隔的字符串 u001.memories：

-- prefix 就是命名空间，用点号连接
SELECT prefix, key FROM store ORDER BY prefix, key;

     prefix     |   key
----------------+---------
 u001.memories  | note-1
 u001.profile   | main
 u002.profile   | main

知道这个格式，就能直接用 SQL 做批量操作了。比如清掉某个用户的全部记忆：

-- 用 LIKE 匹配前缀，删掉这个用户的所有数据
-- 生产环境执行删除前，务必先用 SELECT 确认命中范围
DELETE FROM store WHERE prefix LIKE 'u001.%';

11.5. 完整脚本与输出 #

存成 pg_langgraph.py。这个脚本不调用大模型，所以运行它不花钱、也不需要 API key：

"""在 LangGraph 里用 PostgreSQL：checkpointer 存会话进度，store 存长期记忆。

刻意不调用大模型，所以运行这个脚本不花钱、不需要 API key。
可反复运行，第二次运行会看到上一次的记录还在——这正是要演示的效果。
"""

# 用来给状态字段挂 reducer
import operator
# 类型标注
from typing import Annotated, TypedDict

# 直接连数据库，用来看 LangGraph 到底建了什么表
import psycopg
# PostgreSQL 版的检查点存储：把每一步的状态存进库
from langgraph.checkpoint.postgres import PostgresSaver
# 搭图用的三件套
from langgraph.graph import END, START, StateGraph
# PostgreSQL 版的长期记忆存储：跨会话保存数据
from langgraph.store.postgres import PostgresStore

# 连练习库；sslmode=disable 是本机连接常用写法，避免本地没配证书时报警
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo?sslmode=disable"


# 图的状态：n 是个数字，log 用 operator.add 累加
class State(TypedDict):
    # 普通字段，后写的覆盖先写的
    n: int
    # Annotated 挂上 operator.add，表示新值追加到旧列表后面而不是覆盖
    log: Annotated[list[str], operator.add]


# 图里的第一个节点
def step1(state: State) -> dict:
    """第一步：加一。"""
    # 返回的是「部分更新」，只写要改的字段
    return {"n": state["n"] + 1, "log": ["step1"]}


# 图里的第二个节点
def step2(state: State) -> dict:
    """第二步：乘十。"""
    # n 是覆盖式更新，log 会被追加
    return {"n": state["n"] * 10, "log": ["step2"]}


# 为了让教程输出可复现而加的清理函数
def reset_demo_data() -> None:
    """清掉上一次运行留下的演示数据，让每次运行的输出可复现。

    真实项目里不会这么干，这里只是为了让教程输出稳定。
    """
    # 用普通的 psycopg 连接直接操作这些表
    with psycopg.connect(DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # 先看这些表建了没，没建就不用清
            cur.execute("""
                -- information_schema.tables 里能查到当前库有哪些表
                SELECT table_name FROM information_schema.tables
                -- 只看默认的 public schema
                WHERE table_schema = 'public'
                  -- 只关心 LangGraph 用到的这四张
                  AND table_name IN ('checkpoints', 'checkpoint_blobs', 'checkpoint_writes', 'store')
            """)
            exists = {r[0] for r in cur.fetchall()}
            # checkpointer 的数据分散在三张表里，都按 thread_id 删
            for t in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
                if t in exists:
                    cur.execute(f"DELETE FROM {t} WHERE thread_id LIKE 'demo-thread-%%'")
            # store 的数据按命名空间前缀删
            if "store" in exists:
                cur.execute("DELETE FROM store WHERE prefix LIKE 'u00%%'")
            conn.commit()


def demo_checkpointer() -> None:
    """PostgresSaver：把执行进度存进数据库。"""
    # from_conn_string 是上下文管理器，退出时自动关连接
    with PostgresSaver.from_conn_string(DSN) as saver:
        # 第一次用必须调 setup()，它会建好需要的表；已经建过再调也不报错
        saver.setup()

        # 搭一个最简单的两步图
        g = StateGraph(State)
        g.add_node("step1", step1)
        g.add_node("step2", step2)
        g.add_edge(START, "step1")
        g.add_edge("step1", "step2")
        g.add_edge("step2", END)
        # 编译时把 saver 交给 checkpointer 参数，图就有了持久化能力
        app = g.compile(checkpointer=saver)

        # thread_id 是「这是哪一次会话」的标识，同一个 id 共享历史
        cfg = {"configurable": {"thread_id": "demo-thread-1"}}
        # 跑一次
        out = app.invoke({"n": 1, "log": []}, cfg)
        print("  第一次跑:", out)

        # get_state 把存下来的状态读回来，这是数据库版最直观的好处
        snap = app.get_state(cfg)
        print("  从库里读回的状态:", snap.values)

        # 用同一个 thread_id 再跑：log 会在上一次的基础上继续累加
        out2 = app.invoke({"n": 100, "log": []}, cfg)
        print("  同一 thread 再跑（log 累加）:", out2)

        # 换一个 thread_id：互不干扰，log 从空开始
        out3 = app.invoke({"n": 1, "log": []}, {"configurable": {"thread_id": "demo-thread-2"}})
        print("  换 thread（互不干扰）:", out3)


def demo_store() -> None:
    """PostgresStore：跨会话的长期记忆。"""
    with PostgresStore.from_conn_string(DSN) as store:
        # 同样要先 setup() 建表
        store.setup()

        # put(命名空间, key, 值)；命名空间是元组，习惯上第一段放用户 id
        store.put(("u001", "profile"), "main", {"name": "张三", "city": "杭州"})
        store.put(("u001", "memories"), "note-1", {"text": "喜欢简洁回答"})
        # 换个用户写一条，用来演示隔离
        store.put(("u002", "profile"), "main", {"name": "李四", "city": "北京"})

        # get 按 key 精确取一条，返回的对象用 .value 拿内容
        print("  u001 的档案:", store.get(("u001", "profile"), "main").value)
        # search 按命名空间列出多条
        print("  u001 的笔记:", [(i.key, i.value) for i in store.search(("u001", "memories"))])
        print("  u002 的档案:", store.get(("u002", "profile"), "main").value)
        # 不同命名空间天然隔离，u001 拿不到 u002 的东西
        print("  两个用户的数据互不干扰:", store.get(("u001", "profile"), "main").value["city"])
        # list_namespaces 看现在有哪些命名空间
        print("  所有命名空间:", store.list_namespaces())


# 用普通 SQL 看一眼 LangGraph 在库里留下了什么
def show_tables() -> None:
    """看看上面两个 setup() 到底建了什么表。"""
    # 普通连接
    with psycopg.connect(DSN) as conn:
        # 拿游标
        with conn.cursor() as cur:
            # information_schema.tables 是标准 SQL 的元数据视图
            cur.execute("""
                SELECT table_name FROM information_schema.tables
                WHERE table_schema = 'public'
                  AND (table_name LIKE 'checkpoint%' OR table_name LIKE 'store%')
                ORDER BY table_name
            """)
            # 一共会有 6 张：4 张 checkpoint 的 + 2 张 store 的
            print("  LangGraph 建的表:", [r[0] for r in cur.fetchall()])

            # checkpoints 表里按 thread 数一下有多少条记录
            cur.execute("SELECT thread_id, count(*) FROM checkpoints GROUP BY thread_id ORDER BY thread_id")
            # 跑了几次、每次几步，都能从这个数字看出来
            print("  checkpoints 按 thread 统计:", cur.fetchall())

            # store 表里命名空间是拼成点号字符串存的
            cur.execute("SELECT prefix, key FROM store ORDER BY prefix, key")
            # 可以看到 ("u001", "memories") 变成了 'u001.memories'
            print("  store 表里的记录:", cur.fetchall())


# 直接运行时依次演示
if __name__ == "__main__":
    # 先清掉上次的演示数据；把这一行注释掉再运行，就能看到数据真的留在库里
    reset_demo_data()
    # 第一部分：checkpointer
    print("----- checkpointer：存会话进度 -----")
    # 跑 checkpointer 的演示
    demo_checkpointer()
    # 第二部分：store
    print("\n----- store：存长期记忆 -----")
    # 跑 store 的演示
    demo_store()
    # 第三部分：看库里的表
    print("\n----- 它们在库里建了什么 -----")
    # 用 SQL 查一遍
    show_tables()

实测输出：

----- checkpointer：存会话进度 -----
  第一次跑: {'n': 20, 'log': ['step1', 'step2']}
  从库里读回的状态: {'n': 20, 'log': ['step1', 'step2']}
  同一 thread 再跑（log 累加）: {'n': 1010, 'log': ['step1', 'step2', 'step1', 'step2']}
  换 thread（互不干扰）: {'n': 20, 'log': ['step1', 'step2']}

----- store：存长期记忆 -----
  u001 的档案: {'city': '杭州', 'name': '张三'}
  u001 的笔记: [('note-1', {'text': '喜欢简洁回答'})]
  u002 的档案: {'city': '北京', 'name': '李四'}
  两个用户的数据互不干扰: 杭州
  所有命名空间: [('u001', 'memories'), ('u001', 'profile'), ('u002', 'profile')]

----- 它们在库里建了什么 -----
  LangGraph 建的表: ['checkpoint_blobs', 'checkpoint_migrations', 'checkpoint_writes', 'checkpoints', 'store', 'store_migrations']
  checkpoints 按 thread 统计: [('demo-thread-1', 8), ('demo-thread-2', 4)]
  store 表里的记录: [('u001.memories', 'note-1'), ('u001.profile', 'main'), ('u002.profile', 'main')]

输出里有三处值得对着看：

第一，同一 thread 再跑 那行的 log 有 4 个元素，而 换 thread 只有 2 个。这就是 thread_id 隔离的效果——同一个 thread 接着上次累加，不同 thread 互不影响。

第二，n 从 100 变成 1010（先 +1 得 101，再 ×10 得 1010），说明第二次 invoke 传进去的 n=100 覆盖了旧值。n 没挂 reducer 所以是覆盖，log 挂了 operator.add 所以是累加——同一次运行里两种行为的对比。

第三，checkpoints 里 thread-1 有 8 条记录，thread-2 有 4 条。thread-1 跑了两次、每次 4 个检查点（起点、step1 后、step2 后、终点），所以是 8 条。LangGraph 存的是每一步的快照，不是只存最终结果，这也是它能做「回到上一步重跑」的基础。

最后一个建议：把脚本里的 reset_demo_data() 注释掉再运行一次。 你会看到 第一次跑 的 log 里已经有上次留下的记录了——这才是持久化真正的样子，也是内存版永远做不到的事。

12. 中文环境的两个坑 #

这两个坑跟技术水平无关，纯粹是中文环境带来的，但不知道会白花很多时间。

12.1. 排序：默认是拼音序 #

ORDER BY 中文时，结果取决于数据库的排序规则（collation）。先看本机的设置：

-- datcollate 决定文本怎么比大小，进而决定 ORDER BY 的结果
SELECT datcollate FROM pg_database WHERE datname = current_database();

Chinese (Simplified)_China.936

用五个首字拼音分散的词测一下（阿 a、北 b、李 l、王 w、张 z）：

-- 不加 COLLATE 就用数据库的默认规则
SELECT w FROM words ORDER BY w;

实测结果：

['阿姨', '北京', '李四', '王五', '张三']

这是按拼音排的（a、b、l、w、z），符合中文用户的直觉，很好用。

但如果你（或者某篇教程）为了「让排序稳定」加上了 COLLATE "C"，结果就完全不一样了：

-- COLLATE "C" 表示「按字符编码的字节值排」，和语言无关
SELECT w FROM words ORDER BY w COLLATE "C";

实测结果：

['北京', '张三', '李四', '王五', '阿姨']

这个顺序对中文用户毫无意义——它是按 UTF-8 编码的字节大小排的，跟拼音、笔画、部首都没关系。

所以结论是：

场景	怎么做
给用户看的中文列表	用默认排序（拼音序），别加 COLLATE "C"
需要跨机器完全一致的排序（比如做数据校验）	用 COLLATE "C"，因为它不依赖系统的语言库
建库时	中文项目建议保持中文 collation，不要为了「保险」改成 C

一个容易踩的连带问题： 如果你在 A 机器（中文 collation）导出数据、在 B 机器（英文 collation）导入，排序结果会变。做数据迁移时要留意两边的 collation 是否一致。

12.2. 报错语言与乱码 #

PostgreSQL 的报错会跟随服务端的语言设置。本机是中文环境，所以连上之后的报错都是中文的，这其实挺友好：

# 驱动
import psycopg

# 连接串
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

with psycopg.connect(DSN) as conn:
    with conn.cursor() as cur:
        # 故意查一张不存在的表
        try:
            cur.execute("SELECT * FROM nosuchtable")
        except psycopg.errors.UndefinedTable as e:
            print("默认报错:", str(e).splitlines()[0])
        # 出错后必须 rollback
        conn.rollback()

        # SET lc_messages 可以把本次会话的报错语言切成英文
        # 'C' 表示「不做本地化」，也就是用原始的英文
        cur.execute("SET lc_messages = 'C'")
        try:
            cur.execute("SELECT * FROM nosuchtable")
        except psycopg.errors.UndefinedTable as e:
            print("切成英文后:", str(e).splitlines()[0])
        conn.rollback()

实测输出：

默认报错: 关系 "nosuchtable" 不存在
切成英文后: relation "nosuchtable" does not exist

什么时候需要切成英文？ 搜索报错的时候。中文报错在网上几乎搜不到结果，而英文原文一搜就有大量讨论。所以排查陌生问题时，SET lc_messages = 'C' 拿到英文再去搜，效率高得多。

但有一类报错切不过来，也修不了乱码：连接阶段的报错。 比如密码错了：

connection failed: connection to server at "127.0.0.1", port 5432 failed:
��������:  �û� "postgres" Password ��֤ʧ��

原因是身份验证发生在连接建立之前，那时候客户端和服务端还没协商好用什么编码，服务端按系统本地编码（GBK）发来的中文，客户端按 UTF-8 去解码，就成了乱码。这个阶段还没有「会话」，所以 SET lc_messages 根本没机会执行，在连接串里加 client_encoding=UTF8 也没用——实测过，一样乱码。

应对办法是不看文字，看异常类型，它们是英文的、可靠的：

异常类型	真正的原因
OperationalError + 乱码	密码错、用户不存在，或者库不存在
ConnectionTimeout	服务没启动，或者端口/主机写错了

想进一步区分「密码错」还是「库不存在」，最快的办法是用 psql 手动连一次（先 chcp 65001），它的输出编码是对的。

12.3. 完整脚本与输出 #

把这两个坑合成一个脚本，你可以在自己机器上跑一遍看看结果——如果你的 collation 和本文不同，输出会不一样，这本身就是最值得知道的信息。 存成 pg_chinese.py：

"""中文环境下的两个坑：排序规则和报错语言。

可反复运行，只建一张临时小表。
"""

# 驱动
import psycopg

# 连练习库
DSN = "postgresql://postgres:postgres@localhost:5432/pgdemo"

# 建立连接
with psycopg.connect(DSN) as conn:
    # 拿游标
    with conn.cursor() as cur:
        # ---------- 1. 先看当前库用的是什么排序规则 ----------
        # datcollate 决定文本怎么比大小，进而决定 ORDER BY 的结果
        cur.execute("SELECT datcollate FROM pg_database WHERE datname = current_database()")
        print("当前库的排序规则:", cur.fetchone()[0])

        # ---------- 2. 造几个中文词 ----------
        # 反复运行前先删表
        cur.execute("DROP TABLE IF EXISTS words")
        # 只要一个文本字段就够了
        cur.execute("CREATE TABLE words (w TEXT)")
        # 故意选首字拼音分散的词：阿(a) 北(b) 李(l) 王(w) 张(z)
        cur.executemany(
            "INSERT INTO words (w) VALUES (%s)",
            [("张三",), ("李四",), ("王五",), ("阿姨",), ("北京",)],
        )
        conn.commit()

        # ---------- 3. 默认排序 ----------
        # 不加 COLLATE 就用库的默认规则
        cur.execute("SELECT w FROM words ORDER BY w")
        print("默认排序:      ", [r[0] for r in cur.fetchall()])

        # ---------- 4. 强制按字节排序 ----------
        # COLLATE "C" 表示「按字符编码的字节值排」，和语言无关
        cur.execute('SELECT w FROM words ORDER BY w COLLATE "C"')
        print('COLLATE "C":   ', [r[0] for r in cur.fetchall()])

        # ---------- 5. 报错语言：默认跟随服务端 locale ----------
        # lc_messages 决定报错用什么语言
        cur.execute("SHOW lc_messages")
        print("\n报错语言 lc_messages =", cur.fetchone()[0])

        # 制造一个错，看默认报错长什么样
        try:
            cur.execute("SELECT * FROM nosuchtable")
        except psycopg.errors.UndefinedTable as e:
            print("默认报错:", str(e).splitlines()[0])
        # 出错后必须 rollback
        conn.rollback()

        # 把本次会话的报错语言切成英文
        cur.execute("SET lc_messages = 'C'")
        # 再制造同一个错
        try:
            cur.execute("SELECT * FROM nosuchtable")
        except psycopg.errors.UndefinedTable as e:
            print("切成英文后:", str(e).splitlines()[0])
        conn.rollback()

实测输出：

当前库的排序规则: Chinese (Simplified)_China.936
默认排序:       ['阿姨', '北京', '李四', '王五', '张三']
COLLATE "C":    ['北京', '张三', '李四', '王五', '阿姨']

报错语言 lc_messages = Chinese (Simplified)_China.936
默认报错: 关系 "nosuchtable" 不存在
切成英文后: relation "nosuchtable" does not exist

13. 常见报错对照表 #

遇到的报错八成在这张表里。先看异常类型和关键字，再对照原因。 表里的报错文本都是本机实测的真实输出。

报错	异常类型	原因	怎么解决
用户 "postgres" Password 认证失败	OperationalError	密码错了	检查连接串里的密码；忘了密码要改 pg_hba.conf 重置
数据库 "xxx" 不存在	OperationalError	库名写错，或者还没建	用 psql -U postgres -l 看有哪些库
connection timeout expired	ConnectionTimeout	服务没启动，或端口写错	安装包：Windows 服务里确认 PostgreSQL 在运行；Docker：docker start pg-tutorial；确认端口是 5432
关系 "xxx" 不存在	UndefinedTable	表名写错，或者连错了库	\dt 看当前库有哪些表；确认连接串里的库名
字段 "xxx" 不存在	UndefinedColumn	列名写错	\d 表名 看实际的列名
重复键违反唯一约束	UniqueViolation	插入的值和已有数据重复	先查是否存在，或用 ON CONFLICT 处理冲突
违反了检查约束	CheckViolation	值不满足 CHECK 条件	检查数据，比如年龄是不是填了负数
当前事务被终止, 事务块结束之前的查询被忽略	InFailedSqlTransaction	前面有语句报错但没 rollback	捕获异常后一定要 conn.rollback()（见 7.4 节）
无效的类型 integer 输入语法: "abc"	InvalidTextRepresentation	往数字列里塞了非数字	检查传参；用参数化查询让驱动做类型转换
除以零	DivisionByZero	除数是 0	用 NULLIF(除数, 0) 把 0 变成 NULL，结果会是 NULL 而不是报错
数据插了但查不到	无报错	忘了 commit	写操作之后调 conn.commit()（见 7.3 节）
找不到 psql 命令	无	安装包的 bin 没进 PATH，或走的是 Docker	安装包：把 C:\Program Files\PostgreSQL\18\bin 加到 PATH；Docker：用 docker exec -it pg-tutorial psql -U postgres（见 4.2.2 节）
命令行输出中文乱码	无	Windows 代码页不是 UTF-8	先执行 chcp 65001
无效的 "UTF8" 编码字节顺序	无	用 -c 传了中文 SQL	把 SQL 写进 .sql 文件，用 -f 执行（见 5.3 节）

「忘了 commit」这一条要特别留意，因为它是唯一一个完全不报错的问题。所有语句都成功、rowcount 也对，但数据就是没进去。如果你遇到「明明插入成功了却查不到」，先检查有没有 commit。

14. 和 SQLite / MySQL 怎么选 #

三者都是成熟可靠的选择，没有绝对的优劣，只有适不适合。

先看最实际的决策依据：

你的情况	建议	理由
本地小工具、做实验、单机 App	SQLite	不用装不用配，一个文件就是一个库
做本教程的 LangGraph 持久化	PostgreSQL	官方对 PostgresSaver / PostgresStore 支持最完整
数据结构不固定，要存 JSON	PostgreSQL	JSONB 能建索引，查询性能好
普通 Web 网站，读多写少	两者都行	MySQL 生态和运维资料更多
复杂查询、多表统计、报表	PostgreSQL	窗口函数、CTE 等功能更强
对数据准确性要求极高	PostgreSQL	约束检查更严格，不会静默截断数据

PostgreSQL 相比 MySQL 的关键差别：

	PostgreSQL	MySQL
数据校验	严格，不合规直接报错	某些配置下会静默截断或转换
JSON	JSONB 可建索引，性能好	有 JSON 类型，索引能力较弱
复杂查询	窗口函数、CTE 成熟	8.0 之后才支持，相对简单
连接开销	较高，必须配连接池	较低
大小写	表名默认转小写	依赖操作系统，Windows/Linux 行为不同

「数据校验严格」这一点值得展开说，它是选 PostgreSQL 最实际的理由之一。 往一个 INT 列里插入 'abc'，PostgreSQL 直接报错拒绝；某些 MySQL 配置下会存成 0 并且只给一个警告。报错你会立刻发现并修，静默转换则会让错数据一直躺在库里，几个月后对账才发现。

PostgreSQL 的两个真实短板也说清楚：

1. 连接开销高。 第 10 节实测每个新连接约 50 毫秒，不配连接池的话性能会很难看。这不是可选优化，是必做项。
2. 默认配置偏保守。 装完之后 shared_buffers 只有 128MB，在数据量大的场景需要调，而调优参数确实比 MySQL 复杂一些。不过对学习和中小项目，默认配置完全够用。

15. 小结 #

十条最该带走的结论，按重要性排序：

1. 写完必须 commit。 这是第一名的坑，而且它完全不报错（7.3 节）。
2. 往 SQL 里塞值一律用参数化 %s，永远不要拼字符串。 实测拼字符串会让一个恶意输入把整张表都查出来（7.2 节）。
3. 捕获数据库异常后一定要 rollback。 否则后面所有正常语句都会莫名失败，而报错信息完全指不到真正的错因（7.4 节）。
4. 索引只对「筛得很狠」的查询有效。 实测 60 倍提速，但命中大半张表时优化器会主动放弃它（9.2、9.5 节）。
5. 用 EXPLAIN (ANALYZE) 验证，别靠猜。 看三个东西：扫描方式、Rows Removed by Filter、Execution Time（9.3 节）。
6. 建连接很贵，必须用连接池。 实测每个新连接约 50 毫秒，是一次索引查询耗时的七百倍（10.2 节）。
7. 类型选择有几条硬规矩： 字符串用 TEXT，主键用 BIGINT ... IDENTITY，时间用 TIMESTAMPTZ，钱用 NUMERIC（第 8 节）。
8. 结构不固定的数据用 JSONB 而不是 JSON，因为只有 JSONB 能建索引（8.5 节）。
9. 约束要交给数据库，不要只在代码里检查。 数据库层面的约束谁都绕不过去，是数据不出错的最后一道防线（3.7 节）。
10. Windows 上先 chcp 65001，中文 SQL 写进 .sql 文件用 -f 执行。 能省掉一整类乱码问题（4.4、5.3 节）。

官方文档是最可靠的参考，中文版也在持续维护：PostgreSQL 文档。