PL/Python 基础

PL/pgSQL 是 PostgreSQL 的「母语」，写业务逻辑最顺手，但它在字符串处理、文本解析、数学统计、调用外部库这些场景里编码效率不高。PL/Python 让你在数据库进程内部直接写 Python：函数体就是一段 Python 脚本，参数自动映射成 Python 对象，还能用 plpy 模块执行 SQL。

本页讲清 PL/Python 的安装与定位、语言名为什么只有 plpython3u、数据类型映射规则、plpy 的三种执行方式（execute / prepare / cursor）、SD/GD 会话级共享、触发器里的 TD，以及最关键的安全边界——它是 untrusted 语言，只有超级用户能用。

前置知识

阅读本节前，建议先了解：触发器函数编写

基础概念

PL/Python 是什么

PL/Python 是 PostgreSQL 的可加载过程语言（procedural language）扩展，把 Python 3 解释器嵌入到数据库后端进程里。你在 CREATE FUNCTION 里写的函数体，会被当作文本交给 Python 解释器执行。

text

应用进程                          数据库后端进程（backend）
────────────                      ──────────────────────────────────
SELECT py_cleanup('a {b}');
      │                            ┌─────────────────────────┐
      └──── SQL 请求 ─────────────► │ PostgreSQL 内核          │
                                   │   ↓ 解析、规划、执行      │
                                   │ plpython3u 语言处理器    │
                                   │   ↓ 把函数体编译为        │
                                   │   Python 函数             │
                                   │   ↓                     │
                                   │ CPython 解释器           │
                                   │   ├─ 纯 Python 计算      │
                                   │   ├─ plpy.execute(SQL)   │
                                   │   └─ import 第三方库     │
                                   └─────────────────────────┘

几个直接推论：

推论	说明
无跨进程开销	不像应用层那样要网络往返，适合「大量小计算」
共享数据库内存	Python 对象占用后端进程内存，容易 OOM
阻塞后端	Python 里跑 10 秒，这个连接就 10 秒不响应
GIL 限制	单个后端是单线程 Python，多核用不上
状态不跨会话	每个后端进程独立解释器，会话结束即销毁

只有 `plpython3u` 这一个语言名

这是 PL/Python 最重要的一个事实。官方原文：The language is called plpython3u... A trusted variant plpython might become available in the future.

语言名	是否可用	说明
`plpython3u`	✅ 可用	唯一的 PL/Python 语言名。`u` = untrusted
`plpythonu`	❌	Python 2 时代的名字，PostgreSQL 18 已不再提供
`plpython3`	❌	不存在这个语言名
`plpython`（trusted）	❌ 尚未提供	官方只说「未来可能」，当前不可用

u 后缀（untrusted）意味着什么：函数体可以执行任意 Python 代码，包括 import os; os.system(...)、读写任意文件、发网络请求。因此：

只有超级用户能创建 plpython3u 函数；
普通用户只能调用那些已经被授予 EXECUTE 权限的既有函数。

sql

-- 以普通角色执行会失败
CREATE FUNCTION f() RETURNS int AS $$ return 1 $$ LANGUAGE plpython3u;

text

ERROR:  permission denied for language plpython3u

这与 PL/Perl 形成鲜明对比：PL/Perl 提供了 trusted 的 plperl，PL/Python 没有。所以能用 PL/pgSQL 解决时不要用 PL/Python——一旦引入，权限模型就退化为「超级用户专属」。

安装

PL/Python 是可选的，需要操作系统层面的 Python 3 开发库参与编译。

bash

# Debian / Ubuntu：包名带 Python 版本号
sudo apt-get install postgresql-plpython3-18

# RHEL / Rocky / AlmaLinux
sudo dnf install postgresql18-plpython3

# macOS（Homebrew）
brew install postgresql@18
# Homebrew 官方 formula 通常已启用 plpython3

bash

# 源码编译：需要 --with-python
./configure --with-python
make && make install

安装后在数据库里创建语言：

sql

CREATE EXTENSION plpython3u;
-- 或
CREATE LANGUAGE plpython3u;

验证：

sql

SELECT l.lanname, l.lanpltrusted, p.proname AS handler
FROM pg_language l
JOIN pg_proc p ON p.oid = l.lanplcallfoid
WHERE l.lanname = 'plpython3u';

text

   lanname   | lanpltrusted |      handler
-------------+--------------+--------------------
 plpython3u  | f            | plpython3_call_handler
(1 row)

lanpltrusted = f 正是「untrusted」在系统目录里的体现。

与 PL/pgSQL 的定位对比

维度	PL/pgSQL	PL/Python
是否默认可用	是（内置）	否（需安装扩展）
信任级别	trusted	untrusted
谁能创建	有语言 `USAGE` 权限即可	仅超级用户
数据计算	一般（SQL 表达力强）	强（列表推导、字典、正则、`decimal`）
字符串/文本处理	一般	强（`re`、`json`、`csv`、编解码）
访问文件系统	否	是（`open`、`os`）
调用第三方库	否	是（`import numpy`、`requests`…）
与 SQL 互操作	原生、最自然	通过 `plpy` 桥接
错误处理	`EXCEPTION` 块	`try/except` + `plpy.spiexceptions`
性能（纯计算）	快	中等（类型转换有开销）
性能（大量小调用）	快	慢（每次调用都要转参数）

一个必须提前知道的取舍

把 Python 逻辑放进数据库，通常不是好主意。 它的正当理由只有几条：

场景	是否值得	理由
需要在 SQL 里做复杂文本解析	✅	避免把数据搬到应用层再搬回来
需要在数据库内部调用统计/科学计算库	✅	数据不动，减少传输
需要触发器里做复杂业务判断	⚠️	可以做，但要慎防阻塞
只是想把应用代码搬进数据库	❌	调试难、部署难、权限模型退化
想做 HTTP 调用 / 消息投递	❌	阻塞后端，且失败语义难处理

语法与用法

text

CREATE [ OR REPLACE ] FUNCTION name ( [ argmode ] [ argname ] argtype [, ...] )
    RETURNS rettype
AS $$
    # 这里是 Python 代码
$$ LANGUAGE plpython3u;

最小示例：

sql

CREATE FUNCTION pymax(a integer, b integer)
RETURNS integer
AS $$
if a > b:
    return a
return b
$$ LANGUAGE plpython3u;

text

postgres=# SELECT pymax(3, 7);
 pymax
-------
     7
(1 row)

注意函数体不需要缩进到某个层级——它就是一段独立的 Python 脚本，PostgreSQL 会把它包进一个自动生成的 Python 函数里：

text

你写的函数体                          PostgreSQL 内部转换后
─────────────────────                 ────────────────────────────────────
if a > b:                             def __plpython_procedure_pymax_23456():
    return a                              if a > b:
return b                                      return a
                                          return b
                                      # 23456 是函数的 OID

参数作为全局变量注入，所以能直接用 a、b，也能通过 args 列表访问：

sql

CREATE FUNCTION args_demo(a int, b text) RETURNS text
AS $$
return "args=%r, a=%r, b=%r" % (args, a, b)
$$ LANGUAGE plpython3u;

text

postgres=# SELECT args_demo(1, 'x');
                          args_demo
---------------------------------------------------------------
 args=[1, 'x'], a=1, b='x'
(1 row)

用 SQL 执行：

sql

-- 纯 Python 计算，不碰数据库
SELECT pymax(3, 7);

-- 在查询里当普通函数用
SELECT id, pymax(a, b) FROM t;

-- 作为触发器函数（返回 trigger）
-- 作为过程（CALL）

详细说明

参数是全局变量，不要重新赋值

官方特别提示了一个陷阱：参数被设为全局变量。因为 Python 的作用域规则，在函数体里对同名变量赋值会把它变成局部变量，导致「右侧引用自己」时读到未初始化值。

sql

-- 错误！
CREATE FUNCTION pystrip(x text) RETURNS text
AS $$
x = x.strip()   # UnboundLocalError
return x
$$ LANGUAGE plpython3u;

text

ERROR:  UnboundLocalError: local variable 'x' referenced before assignment

可以用 global 声明修好，但官方建议别依赖这个实现细节：

sql

-- 能跑，但不推荐依赖
CREATE FUNCTION pystrip(x text) RETURNS text
AS $$
global x
x = x.strip()
return x
$$ LANGUAGE plpython3u;

推荐做法：把参数当只读，用新变量承载结果。

sql

-- 推荐：参数只读，结果放新变量
CREATE FUNCTION strip_text(x text) RETURNS text
AS $$
if x is None:
    return None
return x.strip()
$$ LANGUAGE plpython3u;

`None` 与 SQL `NULL`

Python 值	SQL 表现
`None`	`NULL`
未显式 `return`	`None` → `NULL`
`return None`	`NULL`

传入的 SQL NULL 在 Python 里就是 None，必须显式处理，否则会抛 TypeError：

sql

-- 错误！a 为 NULL 时 a > b 抛 TypeError
CREATE FUNCTION bad_max(a integer, b integer) RETURNS integer
AS $$
if a > b:
    return a
return b
$$ LANGUAGE plpython3u;

text

postgres=# SELECT bad_max(NULL, 5);
ERROR:  TypeError: '>' not supported between instances of 'NoneType' and 'int'

两种正确处理方式：

sql

-- 方式一：函数声明 STRICT，NULL 输入直接返回 NULL，不调用 Python
CREATE FUNCTION max_strict(a integer, b integer) RETURNS integer
AS $$
return a if a > b else b
$$ LANGUAGE plpython3u STRICT;

-- 方式二：函数体内显式判空
CREATE FUNCTION max_checked(a integer, b integer) RETURNS integer
AS $$
if a is None or b is None:
    return None
return a if a > b else b
$$ LANGUAGE plpython3u;

STRICT 是 PL/Python 最实用的修饰符：省掉大量判空代码，还能避免不必要的 Python 调用。

数据类型映射表

输入（SQL → Python）：

PostgreSQL 类型	Python 类型
`boolean`	`bool`
`smallint` / `int` / `bigint` / `oid`	`int`
`real` / `double precision`	`float`
`numeric`	`Decimal`（标准库 `decimal`）
`bytea`	`bytes`
数组	`list`（多维为嵌套 `list`）
复合类型	`dict`（字段名为键）
其他（含字符串类型）	`str`

输出（Python → SQL）：

SQL 返回类型	转换规则
`boolean`	按 Python 真值规则：`0`、`''`、`[]`、`{}`、`None` 为假；注意 `'f'` 为真
`bytea`	用 Python 内建转换后写入
其他类型	用 `str()` 转成字符串，再交给该类型的输入函数解析
`float`	用 `repr()` 而非 `str()`，避免精度损失

sql

-- 布尔返回值的经典坑：字符串 'f' 在 Python 里是真值
CREATE FUNCTION bad_bool() RETURNS boolean AS $$
return 'f'      # 非空字符串 → True
$$ LANGUAGE plpython3u;

SELECT bad_bool();

text

 bad_bool
----------
 t
(1 row)

正确写法是返回真正的布尔对象：

sql

CREATE FUNCTION good_bool(x text) RETURNS boolean AS $$
return x == 'f'
$$ LANGUAGE plpython3u;

数组、复合类型与集合返回

sql

-- 数组：list 直通
CREATE FUNCTION return_arr() RETURNS int[]
AS $$
return [1, 2, 3, 4, 5]
$$ LANGUAGE plpython3u;

text

postgres=# SELECT return_arr();
 return_arr
-------------
 {1,2,3,4,5}
(1 row)

字符串是序列，直接返回字符串当数组会得到逐字符拆分——这是个经典陷阱：

sql

-- 错误！
CREATE FUNCTION return_str_arr() RETURNS varchar[]
AS $$
return "hello"
$$ LANGUAGE plpython3u;

text

postgres=# SELECT return_str_arr();
 return_str_arr
----------------
 {h,e,l,l,o}
(1 row)

复合类型输入是 dict：

sql

CREATE TABLE employee(name text, salary integer, age integer);

CREATE FUNCTION overpaid(e employee) RETURNS boolean
AS $$
if e["salary"] > 200000:
    return True
if (e["age"] < 30) and (e["salary"] > 100000):
    return True
return False
$$ LANGUAGE plpython3u;

复合类型输出可用元组/列表或字典：

sql

CREATE TYPE named_value AS (name text, value integer);

-- 用序列返回
CREATE FUNCTION make_pair_seq(name text, value integer) RETURNS named_value
AS $$
return (name, value)
$$ LANGUAGE plpython3u;

-- 用字典返回（键名 = 字段名）
CREATE FUNCTION make_pair_map(name text, value integer) RETURNS named_value
AS $$
return {"name": name, "value": value}
$$ LANGUAGE plpython3u;

集合返回用 yield 最直观：

sql

CREATE TYPE greeting AS (how text, who text);

CREATE FUNCTION greet(how text) RETURNS SETOF greeting
AS $$
for who in ["World", "PostgreSQL", "PL/Python"]:
    yield (how, who)
$$ LANGUAGE plpython3u;

text

postgres=# SELECT * FROM greet('hello');
  how  |    who
-------+------------
 hello | World
 hello | PostgreSQL
 hello | PL/Python
(3 rows)

OUT 参数同样支持：

sql

CREATE FUNCTION multiout_simple(OUT i integer, OUT j integer)
AS $$
return (1, 2)
$$ LANGUAGE plpython3u;

SELECT * FROM multiout_simple();

`plpy.execute`：最简单的 SQL 执行

sql

CREATE FUNCTION count_rows(tbl text) RETURNS bigint
AS $$
rv = plpy.execute("SELECT count(*) AS n FROM " + tbl)
return rv[0]["n"]
$$ LANGUAGE plpython3u;

结果对象 rv 像「列表 + 字典」：

访问方式	说明
`rv[i]`	第 `i` 行（0 基）
`rv[i]["col"]`	第 `i` 行的 `col` 列
`len(rv)`	返回的行数
`rv.nrows()`	命令处理的行数（`UPDATE` 也有值，但返回 0 行）
`rv.status()`	SPI 执行状态字符串
`rv.colnames()`	列名列表
`rv.coltypes()`	列类型 OID 列表
`rv.coltypmods()`	类型修饰符列表

sql

CREATE FUNCTION demo_result_obj() RETURNS text
AS $$
rv = plpy.execute("SELECT 1 AS a, 'x'::text AS b")
return "nrows=%s, len=%s, cols=%s, types=%s" % (
    rv.nrows(), len(rv), rv.colnames(), [str(t) for t in rv.coltypes()])
$$ LANGUAGE plpython3u;

text

postgres=# SELECT demo_result_obj();
                                   demo_result_obj
-------------------------------------------------------------------------------------
 nrows=1, len=1, cols=['a', 'b'], types=['23', '25']
(1 row)

execute 的第二个参数是行数上限，等价于 LIMIT：

sql

CREATE FUNCTION top5() RETURNS int
AS $$
rv = plpy.execute("SELECT id FROM t ORDER BY id", 5)
return len(rv)
$$ LANGUAGE plpython3u;

关键限制（官方明确警告）：plpy.execute 会把整个结果集读进内存。只在你确信结果集不大时使用；大结果集请用 plpy.cursor。

`plpy.prepare` + `plpy.execute`：参数化查询

永远不要用字符串拼接把用户输入塞进 SQL。正确方式是 plpy.prepare：

sql

CREATE FUNCTION find_user(first_name text) RETURNS SETOF text
AS $$
plan = plpy.prepare("SELECT last_name FROM my_users WHERE first_name = $1", ["text"])
rv = plpy.execute(plan, [first_name], 5)
return [r["last_name"] for r in rv]
$$ LANGUAGE plpython3u;

要点：

要点	说明
占位符	`$1`、`$2`…（PostgreSQL 风格，不是 `?`）
类型列表	声明每个占位符的类型，如 `["text", "integer"]`
执行方式	`plpy.execute(plan, [值...], limit)` 或 `plan.execute([值...], limit)`
计划缓存	用 `plpy.prepare` 创建的计划会自动保存，跨调用复用

sql

-- 错误！拼接用户输入 = SQL 注入
rv = plpy.execute("SELECT * FROM users WHERE name = '" + name + "'")

因为计划会被自动缓存，跨函数调用复用时应该存进 SD/GD：

sql

CREATE FUNCTION usesavedplan() RETURNS trigger AS $$
if "plan" in SD:
    plan = SD["plan"]
else:
    plan = plpy.prepare("SELECT 1 AS x")
    SD["plan"] = plan
rv = plpy.execute(plan)
return None
$$ LANGUAGE plpython3u;

`plpy.cursor`：大结果集分块处理

sql

-- 迭代器方式：一次一行
CREATE FUNCTION count_odd_iterator() RETURNS integer AS $$
odd = 0
for row in plpy.cursor("SELECT num FROM largetable"):
    if row["num"] % 2:
        odd += 1
return odd
$$ LANGUAGE plpython3u;

-- fetch 方式：一次一批
CREATE FUNCTION count_odd_fetch(batch_size integer) RETURNS integer AS $$
odd = 0
cursor = plpy.cursor("SELECT num FROM largetable")
while True:
    rows = cursor.fetch(batch_size)
    if not rows:
        break
    for row in rows:
        if row["num"] % 2:
            odd += 1
return odd
$$ LANGUAGE plpython3u;

方式	返回
`for row in cursor`	每行一个 dict
`cursor.fetch(n)`	一个结果对象（最多 n 行）
`cursor.close()`	显式释放资源（否则自动释放）

sql

-- 配合 prepare 使用
CREATE FUNCTION count_odd_prepared() RETURNS integer AS $$
plan = plpy.prepare("SELECT num FROM largetable WHERE num % $1 <> 0", ["integer"])
rows = list(plpy.cursor(plan, [2]))   # 或 list(plan.cursor([2]))
return len(rows)
$$ LANGUAGE plpython3u;

重要提醒（官方 Tip）：plpy.cursor 的对象不是 DB-API 的 cursor，除了名字以外没有任何共同点，不要按 DB-API 的语义去用它。

`SD` 与 `GD`：会话级共享数据

字典	作用域	用途
`SD`	单个函数的多次调用之间	缓存预处理计划、缓存配置、累计状态
`GD`	同一会话内所有 PL/Python 函数共享	全局缓存（谨慎使用）

sql

CREATE FUNCTION cached_cfg() RETURNS text AS $$
if "cfg" not in SD:
    rv = plpy.execute("SELECT value FROM app_config WHERE key = 'mode'")
    SD["cfg"] = rv[0]["value"] if rv else "default"
return SD["cfg"]
$$ LANGUAGE plpython3u;

sql

-- GD：跨函数共享（同一会话内）
CREATE FUNCTION set_context(v text) RETURNS void AS $$
GD["context"] = v
$$ LANGUAGE plpython3u;

CREATE FUNCTION get_context() RETURNS text AS $$
return GD.get("context", "")
$$ LANGUAGE plpython3u;

text

postgres=# SELECT set_context('batch');
 set_context
-------------
(1 row)
postgres=# SELECT get_context();
 get_context
-------------
 batch
(1 row)

要点：

每个函数有独立的执行环境（独立 globals），所以 myfunc 的全局变量在 myfunc2 里不可见——只有 GD 例外。
SD 和 GD 都是会话级的，连接池复用时可能「继承」上一个请求的状态，因此不要在里面存与用户/请求绑定的敏感数据。
GD 官方明确说 use with care——它是全局可变状态，容易造成难查的 bug。

错误处理：`plpy.SPIError` 与 `spiexceptions`

plpy.execute / plpy.prepare 出错时抛 plpy.SPIError 的子类，用 try/except 捕获：

sql

CREATE FUNCTION try_adding_joe() RETURNS text AS $$
try:
    plpy.execute("INSERT INTO users(username) VALUES ('joe')")
except plpy.SPIError:
    return "something went wrong"
else:
    return "Joe added"
$$ LANGUAGE plpython3u;

要按具体错误类型分支，用 plpy.spiexceptions。它把每个 PostgreSQL 错误条件映射成一个异常类，命名规则是「条件名转驼峰」：

SQLSTATE 条件名	异常类
`division_by_zero`	`DivisionByZero`
`unique_violation`	`UniqueViolation`
`foreign_key_violation`	`ForeignKeyViolation`
`fdw_error`	`FdwError`

sql

CREATE FUNCTION insert_fraction(numerator int, denominator int) RETURNS text AS $$
from plpy import spiexceptions
try:
    plan = plpy.prepare("INSERT INTO fractions (frac) VALUES ($1 / $2)", ["int", "int"])
    plpy.execute(plan, [numerator, denominator])
except spiexceptions.DivisionByZero:
    return "denominator cannot equal zero"
except spiexceptions.UniqueViolation:
    return "already have that fraction"
except plpy.SPIError as e:
    return "other error, SQLSTATE %s" % e.sqlstate
else:
    return "fraction inserted"
$$ LANGUAGE plpython3u;

text

postgres=# SELECT insert_fraction(1, 0);
     insert_fraction
------------------------
 denominator cannot equal zero
(1 row)
postgres=# SELECT insert_fraction(1, 2);
  insert_fraction
-------------------
 fraction inserted
(1 row)

因为 spiexceptions 的所有类都继承自 SPIError，所以捕获 SPIError 就能兜住所有数据库错误。

另一种方式是捕获 SPIError 后读 e.sqlstate：

sql

try:
    plpy.execute("...")
except plpy.SPIError as e:
    if e.sqlstate == '23505':      # unique_violation
        return "duplicate"
    raise                          # 其他错误继续抛出

触发器函数：`TD` 字典

PL/Python 写触发器时，函数返回类型是 trigger，上下文在 TD 字典里：

键	内容
`TD["event"]`	`INSERT` / `UPDATE` / `DELETE` / `TRUNCATE`
`TD["when"]`	`BEFORE` / `AFTER` / `INSTEAD OF`
`TD["level"]`	`ROW` / `STATEMENT`
`TD["new"]`	行级触发器的新行（dict）
`TD["old"]`	行级触发器的旧行（dict）
`TD["name"]`	触发器名
`TD["table_name"]`	表名
`TD["table_schema"]`	表所在 schema
`TD["relid"]`	表 OID
`TD["args"]`	`CREATE TRIGGER` 里传入的参数，`TD["args"][0]` 起

行级 BEFORE / INSTEAD OF 触发器的返回值语义（官方原文规则）：

返回	效果
`None`	该行不做修改（与原行相同）
`"OK"`	同上
`"SKIP"`	跳过该行操作
`"MODIFY"`	你修改了 `TD["new"]`，请用新值（仅 `INSERT` / `UPDATE`）

其他情况的返回值被忽略。

sql

CREATE FUNCTION py_audit_user() RETURNS trigger AS $$
import json
plan = plpy.prepare(
    "INSERT INTO user_audit(user_id, action, old_row, new_row, at) "
    "VALUES ($1, $2, $3, $4, now())",
    ["bigint", "text", "jsonb", "jsonb"])
plpy.execute(plan, [
    (TD["new"] or TD["old"])["id"],
    TD["event"],
    json.dumps(TD["old"]) if TD["old"] else None,
    json.dumps(TD["new"]) if TD["new"] else None,
])
return None    # AFTER 触发器，返回值被忽略
$$ LANGUAGE plpython3u;

CREATE TRIGGER trg_py_audit_user
    AFTER INSERT OR UPDATE OR DELETE ON users
    FOR EACH ROW EXECUTE FUNCTION py_audit_user();

一个「改值」的 BEFORE 触发器：

sql

CREATE FUNCTION py_normalize_email() RETURNS trigger AS $$
new = TD["new"]
new["email"] = new["email"].strip().lower()
return "MODIFY"     # 告诉 PostgreSQL：我改了 new 行，请用新值
$$ LANGUAGE plpython3u;

CREATE TRIGGER trg_norm_email
    BEFORE INSERT OR UPDATE ON users
    FOR EACH ROW EXECUTE FUNCTION py_normalize_email();

匿名代码块：`DO` 语句

PL/Python 支持在 DO 块里直接跑一段临时 Python：

sql

DO $$
import csv, io
rv = plpy.execute("SELECT id, name FROM t")
buf = io.StringIO()
w = csv.writer(buf)
w.writerow(["id", "name"])
for r in rv:
    w.writerow([r["id"], r["name"]])
plpy.notice(buf.getvalue())
$$ LANGUAGE plpython3u;

注意 DO 块的函数体规则与函数一致（返回 None），且同样受超级用户权限约束。

显式子事务

PL/Python 可以在 Python 里做显式子事务边界（对应 PL/pgSQL 的 EXCEPTION 块），API 是 plpy.subtransaction()，作为上下文管理器使用：

sql

CREATE FUNCTION try_insert() RETURNS text AS $$
try:
    with plpy.subtransaction():
        plpy.execute("INSERT INTO t(id) VALUES (1)")
        plpy.execute("INSERT INTO t(id) VALUES (1)")   # 唯一冲突
except plpy.SPIError as e:
    return "rolled back subtransaction: SQLSTATE=%s" % e.sqlstate
return "committed"
$$ LANGUAGE plpython3u;

用它可以把「一组语句」包成原子单元，出错时只回滚这个单元，外层事务继续。

实战示例

示例一：用 Python 做复杂文本清洗

sql

CREATE EXTENSION IF NOT EXISTS plpython3u;

CREATE FUNCTION clean_phone(raw text) RETURNS text
AS $$
import re
if raw is None:
    return None
digits = re.sub(r"\D", "", raw)
if len(digits) == 11 and digits.startswith("1"):
    return digits
if len(digits) == 13 and digits.startswith("86"):
    return digits[2:]
return None
$$ LANGUAGE plpython3u IMMUTABLE STRICT;

text

postgres=# SELECT clean_phone('+86 138-0013-8000') AS a,
                  clean_phone('(010) 8888 6666') AS b;
       a       |  b
---------------+------
 13800138000   |
(1 row)

必须加 IMMUTABLE：函数对同样输入永远返回同样结果，规划器才能用它建表达式索引。

sql

CREATE INDEX idx_users_phone_clean ON users (clean_phone(phone_raw));

示例二：用 Python 解析半结构化日志

sql

CREATE FUNCTION parse_log_line(line text) RETURNS jsonb
AS $$
import re, json
if line is None:
    return None
m = re.match(r'^(?P<ts>\S+ \S+) \[(?P<lvl>\w+)\] (?P<svc>[\w\-]+): (?P<msg>.*)$', line)
if not m:
    return json.dumps({"raw": line, "parsed": False})
return json.dumps({
    "ts": m.group("ts"),
    "level": m.group("level"),
    "service": m.group("svc"),
    "message": m.group("msg"),
    "parsed": True,
})
$$ LANGUAGE plpython3u IMMUTABLE STRICT;

text

postgres=# SELECT parse_log_line('2026-09-28 10:01:02 [ERROR] payment: timeout after 5s');
                                    parse_log_line
------------------------------------------------------------------------------------
 {"ts": "2026-09-28 10:01:02", "level": "ERROR", "service": "payment", "message": "timeout after 5s", "parsed": true}
(1 row)

注意返回类型是 jsonb：Python 侧用 json.dumps() 生成字符串，PostgreSQL 侧按 jsonb 输入函数解析。不要返回 dict——jsonb 的输入函数期待的是 JSON 文本。

示例三：数据质量校验触发器

sql

CREATE TABLE dq_rules (
    table_name text NOT NULL,
    column_name text NOT NULL,
    rule_type  text NOT NULL,      -- not_null / regex / range
    param      text,
    PRIMARY KEY (table_name, column_name, rule_type)
);

CREATE FUNCTION py_dq_check() RETURNS trigger
AS $$
import re
row = TD["new"] or TD["old"]
rv = plpy.execute(plpy.prepare(
    "SELECT rule_type, param FROM dq_rules WHERE table_name = $1",
    ["text"]), [TD["table_name"]])
for rule in rv:
    col = rule["column_name"] if "column_name" in rule else None
    val = row.get(col)
    if rule["rule_type"] == "not_null" and val is None:
        plpy.error("字段 %s 不能为空" % col)
    elif rule["rule_type"] == "regex" and val is not None:
        if not re.search(rule["param"], str(val)):
            plpy.error("字段 %s 不匹配规则 %s" % (col, rule["param"]))
return None
$$ LANGUAGE plpython3u;

plpy.error(msg) 会抛出数据库错误（等价 RAISE EXCEPTION），中止当前语句并回滚。同族函数：

函数	对应 SQL 级别
`plpy.debug(msg)`	`DEBUG`
`plpy.log(msg)`	`LOG`
`plpy.info(msg)`	`INFO`
`plpy.notice(msg)`	`NOTICE`
`plpy.warning(msg)`	`WARNING`
`plpy.error(msg)`	`ERROR`（抛异常）
`plpy.fatal(msg)`	`FATAL`（终止会话）

text

postgres=# INSERT INTO users(email) VALUES ('not-an-email');
ERROR:  字段 email 不匹配规则 ^[^@]+@[^@]+\.[^@]+$
CONTEXT:  PL/Python function "py_dq_check"

示例四：分块处理大结果集

sql

CREATE FUNCTION rehash_ids(batch integer DEFAULT 10000) RETURNS bigint
AS $$
total = 0
cursor = plpy.cursor("SELECT id, payload FROM big_table WHERE hash IS NULL")
while True:
    rows = cursor.fetch(batch)
    if not rows:
        break
    plan = plpy.prepare("UPDATE big_table SET hash = $1 WHERE id = $2",
                        ["text", "bigint"])
    for row in rows:
        import hashlib
        h = hashlib.sha256(row["payload"].encode()).hexdigest()
        plpy.execute(plan, [h, row["id"]])
        total += 1
return total
$$ LANGUAGE plpython3u;

text

postgres=# SELECT rehash_ids(1000);
 rehash_ids
------------
     148329
(1 row)

关键设计：

点	说明
`plpy.cursor` 而非 `plpy.execute`	避免把整表读进内存
`plpy.prepare` 复用计划	循环内每次 `prepare` 是浪费
批次大小可控	内存占用与单条开销之间取平衡
单函数单事务	整段是一个事务，失败全回滚

注意：整段是一个事务，回滚会撤销所有 UPDATE。若要分批提交，PL/Python 里做不到（函数内不能控制事务），需要外部循环驱动。

示例五：用 `SD` 缓存配置，避免重复查询

sql

CREATE FUNCTION get_rate(currency text) RETURNS numeric
AS $$
if "rates" not in SD:
    rv = plpy.execute("SELECT code, rate FROM fx_rates")
    SD["rates"] = {r["code"]: r["rate"] for r in rv}
    SD["loaded_at"] = __import__("time").time()

# 每小时刷新一次
import time
if time.time() - SD["loaded_at"] > 3600:
    SD.pop("rates")
    SD.pop("loaded_at")
    return get_rate(currency)

return SD["rates"].get(currency, 0)
$$ LANGUAGE plpython3u;

text

postgres=# SELECT get_rate('USD'), get_rate('EUR'), get_rate('USD');
 get_rate | get_rate | get_rate
----------+----------+----------
     7.12 |     7.75 |     7.12
(1 row)
-- 后两次调用直接命中 SD 缓存，没有查库

这是 SD 最典型也最有价值的用法：把「每行调用一次函数都要查一次配置表」变成「每会话查一次」。

示例六：调用 Python 生态做校验

sql

CREATE FUNCTION is_valid_iban(iban text) RETURNS boolean
AS $$
# 仅供演示：真实校验请用专门的库
if iban is None:
    return False
s = iban.replace(" ", "").upper()
if len(s) < 15 or not s[:2].isalpha() or not s[2:4].isdigit():
    return False
rearranged = s[4:] + s[:4]
digits = "".join(str(int(c, 36)) for c in rearranged)
return int(digits) % 97 == 1
$$ LANGUAGE plpython3u IMMUTABLE STRICT;

text

postgres=# SELECT is_valid_iban('GB82 WEST 1234 5698 7654 32');
 is_valid_iban
---------------
 t
(1 row)

这个示例的意义在于展示能 import 任意 Python 库——包括 numpy、pandas、cryptography 等。但要注意：装库要装在数据库服务器上，且要与服务器使用的 Python 版本一致。

bash

# 必须装到数据库服务器用的那个 Python 环境
/usr/bin/python3 -m pip install numpy
# 验证
psql -c "DO \$\$ import numpy; plpy.notice(numpy.__version__) \$\$ LANGUAGE plpython3u;"

注意事项

不要以为普通用户可以创建 plpython3u 函数。 它是 untrusted 语言，创建需要超级用户。

sql

-- 错误！
-- 普通角色执行会得到：permission denied for language plpython3u
CREATE FUNCTION f() RETURNS int AS $$ return 1 $$ LANGUAGE plpython3u;

不要用字符串拼接把参数塞进 SQL。 必须用 plpy.prepare 的 $n 占位符。

sql

-- 错误！SQL 注入
rv = plpy.execute("SELECT * FROM users WHERE name = '" + name + "'")

不要在函数体里给参数重新赋值。 参数是全局变量，赋值会触发 UnboundLocalError。
sql
```
-- 错误！
x = x.strip()
```
不要用 plpy.execute 处理大结果集。 它把整个结果读进内存，改用 plpy.cursor。
不要忘记判空。 除非函数声明了 STRICT，否则 NULL 会以 None 传进来，参与运算就抛 TypeError。
不要在返回 boolean 时返回字符串。 Python 真值规则下 'f' 是 True。
sql
```
-- 错误！返回 'f' 得到 true
return 'f'
```

不要把字符串当数组返回。 字符串是序列，会被逐字符拆开。

sql

-- 错误！结果变成 {h,e,l,l,o}
return "hello"   -- 而函数声明 RETURNS varchar[]

不要在 GD 里存与用户或请求绑定的数据。 GD 是会话级共享，连接池复用时会串味。
不要在 PL/Python 里执行长耗时操作。 它阻塞整个后端进程，连接池会被拖垮，且无法中途取消（除非从外部 pg_cancel_backend）。
不要在函数里试图 COMMIT。 函数运行在调用者的事务里，PL/Python 没有事务控制接口（只有 plpy.subtransaction() 做子事务）。
不要假设 PL/Python 一定已安装。 它是可选组件，CREATE EXTENSION plpython3u 会因缺少 .so 而失败。
text
```
ERROR:  could not open extension control file ".../plpython3u.control"
```
不要在 plpython3u 里做外部网络 I/O 并依赖它成功。 这是最常见的「把数据库当应用服务器」反模式：失败语义难处理、超时会拖死连接。

不要忘记 IMMUTABLE / STABLE / VOLATILE 标记。 影响规划器能否用表达式索引、能否在索引条件里下推。

sql

-- 不推荐：纯函数却标默认 VOLATILE，无法用于表达式索引
CREATE FUNCTION f(x text) RETURNS text AS $$ ... $$ LANGUAGE plpython3u;

不要忽略 Python 版本差异。 服务器上的 Python 版本决定了你能用哪些语法（如 match 语句需 3.10+）。

最佳实践

优先用 PL/pgSQL，只在「Python 明显更擅长」时才用 PL/Python。 权限模型与可维护性都是成本。

能用 STRICT 就用 STRICT。 省掉判空代码，还避免无谓的 Python 调用。

sql

CREATE FUNCTION slugify(t text) RETURNS text
AS $$ ... $$ LANGUAGE plpython3u IMMUTABLE STRICT;

修正确定性的纯函数要标 IMMUTABLE，才能建表达式索引。
sql
```
CREATE INDEX idx_u_slug ON users (slugify(display_name));
```

所有含参数的 SQL 都用 plpy.prepare + $n。

python

plan = plpy.prepare("SELECT * FROM t WHERE a = $1 AND b = $2", ["int", "text"])
rv = plpy.execute(plan, [a, b])

把预处理计划缓存在 SD 里，避免重复 prepare。

python

if "plan" not in SD:
    SD["plan"] = plpy.prepare("SELECT ...")
rv = plpy.execute(SD["plan"], [x])

大结果集一律用 plpy.cursor，并显式设定批量大小。

python

cursor = plpy.cursor("SELECT ... FROM big_table")
while True:
    rows = cursor.fetch(10000)
    if not rows:
        break

错误处理优先用 plpy.spiexceptions 的具名异常，最后用 plpy.SPIError 兜底。

python

from plpy import spiexceptions
try:
    ...
except spiexceptions.UniqueViolation:
    ...
except plpy.SPIError as e:
    plpy.warning("unexpected SQLSTATE %s", e.sqlstate)
    raise

SD 只缓存「与请求无关」的只读数据（配置、外币汇率、字典映射），并考虑刷新策略。
把日志输出用 plpy.notice / plpy.warning，不要用 print。 print 的输出不会进数据库日志。
python
```
# 不推荐
print("debug:", value)
# 推荐
plpy.notice("value=%s", value)
```
审计/调试用的 plpy.debug 需先把 log_min_messages 调到 debug1 才可见。

给 PL/Python 函数单独放一个 schema，并严格控制 EXECUTE 授权。

sql

REVOKE ALL ON FUNCTION py.tool(text) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION py.tool(text) TO app_writer;

上线前在服务器上验证所需的 Python 库已安装且版本正确。

bash

psql -Atc "DO \$\$ import numpy; plpy.notice(numpy.__version__) \$\$ LANGUAGE plpython3u;"

把「数据库服务器上的 Python 库清单」纳入部署文档与镜像构建。 这是最容易在迁移时踩的坑。

下一节：PL/Perl 基础——另一种嵌入式过程语言，它提供了 trusted 变体，权限模型与 PL/Python 完全不同。

PL/Python 基础 ​

基础概念 ​

PL/Python 是什么 ​

只有 plpython3u 这一个语言名 ​

安装 ​

与 PL/pgSQL 的定位对比 ​

一个必须提前知道的取舍 ​

语法与用法 ​

详细说明 ​

参数是全局变量，不要重新赋值 ​

None 与 SQL NULL ​

数据类型映射表 ​

数组、复合类型与集合返回 ​

plpy.execute：最简单的 SQL 执行 ​

plpy.prepare + plpy.execute：参数化查询 ​

plpy.cursor：大结果集分块处理 ​

SD 与 GD：会话级共享数据 ​

错误处理：plpy.SPIError 与 spiexceptions ​

触发器函数：TD 字典 ​

匿名代码块：DO 语句 ​

显式子事务 ​

实战示例 ​

示例一：用 Python 做复杂文本清洗 ​

PL/Python 基础

基础概念

PL/Python 是什么

只有 `plpython3u` 这一个语言名

安装

与 PL/pgSQL 的定位对比

一个必须提前知道的取舍

语法与用法

详细说明

参数是全局变量，不要重新赋值

`None` 与 SQL `NULL`

数据类型映射表

数组、复合类型与集合返回

`plpy.execute`：最简单的 SQL 执行

`plpy.prepare` + `plpy.execute`：参数化查询

`plpy.cursor`：大结果集分块处理

`SD` 与 `GD`：会话级共享数据

错误处理：`plpy.SPIError` 与 `spiexceptions`

触发器函数：`TD` 字典

匿名代码块：`DO` 语句

显式子事务

实战示例

示例一：用 Python 做复杂文本清洗