Python 连接 PostgreSQL

Python 生态里连接 PostgreSQL 的主流选择有四个：psycopg2（二十年历史的经典库，稳定且无处不在）、psycopg3（同一团队的后继者，现代化 API、COPY 与流水线原生支持）、asyncpg（asyncio 时代的性能标杆）、以及 SQLAlchemy（在上述驱动之上提供连接串统一与 ORM/Toolkit 能力）。本页按「先 psycopg2 打地基、再 psycopg3 看演进、后 asyncpg 补异步」的顺序展开，最后用 ETL 实战把 COPY、批量写入、事务串成一个完整脚本。

Python 驱动层有几个与语言特性强绑定的共性：参数占位符是 %s（psycopg 系，注意它不是字符串格式化，传参永远是元组）；事务模型默认自动开启（第一条语句隐式 BEGIN，直到 commit/rollback）；以及 JSONB、数组、UUID 这类 PostgreSQL 特色类型需要适配器或新驱动的原生支持。理解这三点，四个库就能触类旁通。

选型速览：存量代码与依赖生态 → psycopg2；新项目 → psycopg3（或按团队现状）；纯 asyncio 服务 → asyncpg 或 psycopg3 异步模式；需要 ORM/多库支持 → SQLAlchemy 串起底层驱动。

前置知识

建议先阅读：PHP 连接 PostgreSQL

基础概念

库	安装	同步/异步	定位
psycopg2	pip install psycopg2-binary	同步	经典事实标准，生态兼容性最好
psycopg3	pip install "psycopg[binary]"	同步 + 异步	现代后继者，COPY/pipeline 原生
asyncpg	pip install asyncpg	异步	极致性能，API 风格独立
SQLAlchemy	pip install sqlalchemy psycopg2-binary	两者	连接串统一 + ORM/Core

三个贯穿全页的心智模型：

1. 占位符是 %s，参数是元组。cursor.execute(sql, params) 里的 %s 由驱动做安全转义（在 psycopg2 里是客户端转义，在 psycopg3 里走服务端绑定），绝不要用 % 字符串格式化把值拼进 SQL；
2. 事务默认自动开启。psycopg2 从第一条语句起就在事务里，不 commit 就不落盘；psycopg3 延续此设计；asyncpg 则更原始——它不开隐式事务，但每条语句默认在自动提交模式下执行，显式事务用 async with conn.transaction()；
3. 连接是重资源。PostgreSQL 每连接一个服务端进程，Web 服务务必配连接池（驱动内置池或外部 PgBouncer）。

                    ┌──────────────────────────────┐
   Web (FastAPI) ──►│ asyncpg / psycopg3 AsyncPool │──► PostgreSQL
                    └──────────────────────────────┘
                    ┌──────────────────────────────┐
   ETL / 脚本      ──►│ psycopg2 / psycopg3 + COPY   │──► PostgreSQL
                    └──────────────────────────────┘
                    ┌──────────────────────────────┐
   业务应用        ──►│ SQLAlchemy(ORM) → 底层驱动    │──► PostgreSQL
                    └──────────────────────────────┘

语法与用法

psycopg2 最小可用样板：

import psycopg2

conn = psycopg2.connect(
    host="db1.prod.internal", port=5432, dbname="appdb",
    user="app_rw", password="secret", connect_timeout=5, sslmode="require",
)
conn.autocommit = False              # 默认就是 False，显式声明更清晰

with conn.cursor() as cur:
    cur.execute(
        "SELECT id, name FROM users WHERE status = %s AND age > %s",
        ("active", 18),
    )
    rows = cur.fetchall()

conn.commit()        # 读操作也要提交（结束只读事务），否则连接挂着事务
conn.close()

psycopg3 等价写法：

import psycopg

with psycopg.connect(
    "host=db1.prod.internal port=5432 dbname=appdb user=app_rw password=secret",
    connect_timeout=5, sslmode="require",
) as conn:
    conn.execute(
        "SELECT id, name FROM users WHERE status = %s", ("active",)
    )
    rows = conn.fetchall()
# 退出 with 自动提交（无异常）或回滚（有异常）并关闭连接

批量插入与 COPY（psycopg3）：

with psycopg.connect(dsn) as conn, conn.cursor() as cur:
    # COPY：从文件流式导入
    with cur.copy("COPY stg_events FROM STDIN WITH (FORMAT csv, HEADER true)") as cp:
        with open("events.csv", "rb") as f:
            cp.write(f.read())

连接池（psycopg3）：

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    "host=localhost dbname=appdb", min_size=4, max_size=20, open=True,
)
with pool.connection() as conn:
    conn.execute("SELECT 1")

详细说明

psycopg2：连接、游标与 %s 占位符

psycopg2.connect() 接受 keyword 参数或一个 libpq 风格的 DSN 字符串，参数全集与 libpq 一致（connect_timeout、sslmode、application_name 都可用）。游标（cursor）是执行与取数的载体：

cur.execute(sql, params)      # 执行，参数永远走元组
cur.fetchone()                # 单行（无更多行时 None）
cur.fetchmany(size)           # N 行
cur.fetchall()                # 全部行（大结果集慎用）
cur.rowcount                  # 受影响/返回的行数
cur.description               # 列元信息

%s 的语义必须刻进肌肉记忆：它是占位符，不是 Python 的 % 格式化。任何 sql % value、f-string 拼接用户输入都是注入事故。同理，IN 列表要展开成等量占位符：

ids = [1, 2, 3]
cur.execute("SELECT * FROM t WHERE id = ANY(%s)", (ids,))   # 推荐：数组占位符

execute 之后游标状态被占用，连续两次 execute 前若还要取上一次结果，用 fetchall() 缓存；psycopg3 则允许同游标多次 execute。

psycopg2：事务模型与 autocommit

psycopg2 默认第一条语句就隐式开启事务，conn.commit() 提交、conn.rollback() 回滚，conn.autocommit = True 切换到自动提交（DDL、COPY 前导、维护操作常用）。三个工程要点：

# 1) with conn：成功提交、异常回滚（但 conn 由你负责关闭）
with conn:
    with conn.cursor() as cur:
        cur.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (100, 1))

# 2) 长事务要收口：SELECT 之后不 commit，事务一直挂着
cur.execute("SELECT pg_backend_pid()")

# 3) 隔离级别与只读事务
conn.set_session(isolation_level="REPEATABLE READ", readonly=True)

「只读脚本忘记 commit」是 psycopg2 最常见的坑：查询一样在事务里，长时间挂着会阻止 vacuum、给复制添堵。要么 autocommit=True，要么用完即 commit。

psycopg2：取数与行字典化

默认返回元组，按列位置取值；RealDictCursor 让每行变成字典：

from psycopg2.extras import RealDictCursor

with conn.cursor(cursor_factory=RealDictCursor) as cur:
    cur.execute("SELECT id, name, created_at FROM users LIMIT 3")
    rows = cur.fetchall()
print(rows[0]["name"])

还有 DictRow（字典化但保持列序）与 NamedTupleCursor（属性访问 + 类型提示友好）。行工厂的选择影响 downstream 的可读性——API 序列化用 RealDictCursor，数据处理管道用 NamedTupleCursor。

psycopg2：批量写入的量级差距

psycopg2 的 executemany() 名不副实：它只是循环执行，还比手写循环多一次解析，性能与逐条 INSERT 无本质差别。真正快的是 execute_values()——把多行合并成一条多值 INSERT：

from psycopg2.extras import execute_values

rows = [(i, f"name-{i}", i * 1.5) for i in range(100_000)]
execute_values(
    cur,
    "INSERT INTO items (id, name, score) VALUES %s",
    rows,
    template="(%s, %s, %s)",
    page_size=1000,          # 每批行数
)

经验量级：逐条 execute < executemany << execute_values（十倍级差距）< COPY（再快数倍）。万行以下 execute_values 足够，十万行以上直接 COPY。配合 psycopg2.extras.Json 适配器，dict/list 可以直接作为 JSONB 参数：

from psycopg2.extras import Json
cur.execute(
    "INSERT INTO events (payload) VALUES (%s)",
    (Json({"kind": "click", "ts": 1780000000}),),
)

psycopg2：服务端游标与内置连接池

默认游标把全部结果缓存在客户端；几百万行的查询会撑爆内存。命名游标（named cursor）变成服务端游标，按块取回：

with conn.cursor(name="big_query_cur") as cur:   # 有 name 即为服务端游标
    cur.itersize = 10000                          # 每批行数
    cur.execute("SELECT * FROM big_table")
    for row in cur:                               # 迭代即分块拉取
        process(row)

连接池用 psycopg2.pool：SimpleConnectionPool（单线程）与 ThreadedConnectionPool（多线程/多协程 worker）：

from psycopg2.pool import ThreadedConnectionPool

pool = ThreadedConnectionPool(minconn=2, maxconn=20, dsn="host=… dbname=…")
conn = pool.getconn()
try:
    with conn.cursor() as cur:
        cur.execute("SELECT 1")
    conn.commit()
finally:
    pool.putconn(conn)

注意 psycopg2 的池非常朴素（无健康检查、无超时），生产级场景更推荐外部 PgBouncer 或 psycopg3 的池。

psycopg3：连接与执行的新面貌

psycopg3 的安装推荐带二进制包：pip install "psycopg[binary]"。API 的核心变化是连接与游标都支持上下文管理器，且 conn.execute() 短路了游标样板：

import psycopg

with psycopg.connect(dsn) as conn:            # 异常自动回滚，正常退出自动提交
    conn.execute(
        "INSERT INTO items (id, name) VALUES (%s, %s)",
        (1, "x"),
    )
    conn.execute("SELECT count(*) FROM items")
    (cnt,) = conn.fetchone()

# 多行插入：第四参数就是序列，驱动自动展开多值
conn.executemany(
    "INSERT INTO items (id, name) VALUES (%s, %s)",
    [(1, "a"), (2, "b"), (3, "c")],
)

psycopg3 默认走扩展查询协议（服务端绑定），并且对支持的类型自动使用二进制传输（解析更快、网络更省）——这是它比 psycopg2 快的重要原因之一。类型映射也更原生：jsonb 直接映射 dict/list、uuid 映射 uuid.UUID、数组映射 list，不需要适配器。

psycopg3：copy 上下文与 pipeline 模式

cur.copy() 把 COPY 协议封装成上下文管理器，是 Python 侧最快的数据搬运口：

# 导入：逐行/逐块写入
with cur.copy("COPY stg_events FROM STDIN WITH (FORMAT csv, HEADER true)") as cp:
    with open("events.csv", "r") as f:
        for line in f:
            cp.write(line)

# 带类型转换的导入（把每行元素按列类型序列化）
with cur.copy("COPY items (id, name) FROM STDIN") as cp:
    cp.set_types(["int4", "text"])
    for row in rows:
        cp.write_row(row)

# 导出：流式读取
with cur.copy("COPY items TO STDOUT WITH (FORMAT csv)") as cp:
    for chunk in cp:          # 惰性产出，内存平稳
        out.write(chunk)

pipeline() 模式则把多条语句攒批发送，隐藏网络往返延迟——高延迟链路上的一批小查询能快数倍：

with psycopg.connect(dsn) as conn:
    with conn.pipeline():
        for sql, params in jobs:
            conn.execute(sql, params)
    # 退出 pipeline 时统一等待结果

psycopg3：连接池与异步

psycopg_pool 提供生产级池（健康检查、自动重连、超时）：

from psycopg_pool import ConnectionPool, AsyncConnectionPool

pool = ConnectionPool(dsn, min_size=4, max_size=20, open=True)
with pool.connection() as conn:
    conn.execute("SELECT 1")

异步世界的一等公民：

import asyncio
import psycopg
from psycopg_pool import AsyncConnectionPool

async def main():
    pool = AsyncConnectionPool(dsn, min_size=2, max_size=10, open=True)
    async with pool.connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT now()")
            row = await cur.fetchone()
    await asyncio.sleep(0)

asyncio.run(main())

psycopg3 的异步 API 与同步 API 几乎镜像，团队不需要维护两套心智；这是它相对 asyncpg 的最大工程优势。

psycopg3：prepare_threshold 与语句复用

psycopg3 内建了自动预处理：同一条 SQL（文本完全一致）执行满 conn.prepare_threshold 次（默认 5）后，驱动自动改用命名预处理语句复用服务端计划：

with psycopg.connect(dsn) as conn:
    conn.prepare_threshold = 0    # 0 = 首次执行即预处理（调试/基准用）
    for params in hot_jobs:
        conn.execute("UPDATE items SET status = %s WHERE id = %s", params)

阈值保持默认即可——低频语句走扩展查询协议、高频语句自动转预处理，驱动已经做了正确的分档。prepare_threshold = 0 用于基准测试与「确认预处理路径无回归」的排查；与 PgBouncer transaction 模式的兼容性同 PHP / Node 的小节：池在中间时优先让池侧负责语句状态，或评估 QueryExecMode 形态的等价做法。

asyncpg：事件循环时代的性能标杆

asyncpg 只做异步、只支持 asyncio，但性能（尤其批量与 COPY）常年处于第一梯队。它的 API 风格独立：参数占位符是 $1（服务端原生记法）、返回 Record 对象（既支持下标也支持属性访问）、没有游标抽象（取数用 fetch/fetchrow/迭代）：

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect("postgresql://app_rw:secret@db1/appdb")
    rows = await conn.fetch(
        "SELECT id, name FROM users WHERE status = $1 AND age > $2",
        "active", 18,
    )
    print(rows[0]["name"])          # Record 支持 dict 风格访问
    await conn.close()

asyncio.run(main())

批量与池：

# execute_many 批量写入（本质是 COPY 层面的优化）
await conn.executemany("INSERT INTO items (id, name) VALUES ($1, $2)", rows)

# 内置池
pool = await asyncpg.create_pool(dsn, min_size=2, max_size=10)
async with pool.acquire() as conn:
    rows = await conn.fetch("SELECT 1")

# 显式事务
async with conn.transaction():
    await conn.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", 100, 1)

asyncpg 不自动开启隐式事务、SQLSTATE 通过 asyncpg.UniqueViolationError 等异常类表达（也有 e.sqlstate 属性），错误处理按异常类型分支。

SQLAlchemy 连接串与 JSONB 写法

SQLAlchemy 的 PostgreSQL 方言 URL 决定底层驱动：

# psycopg2（默认）
engine = create_engine("postgresql+psycopg2://app_rw:secret@db1:5432/appdb")
# psycopg3
engine = create_engine("postgresql+psycopg://app_rw:secret@db1:5432/appdb")
# asyncpg（异步引擎）
engine = create_async_engine("postgresql+asyncpg://app_rw:secret@db1:5432/appdb")

# 连接参数挂在 URL 之后
engine = create_engine(
    "postgresql+psycopg2://app_rw:secret@db1:5432/appdb"
    "?connect_timeout=5&sslmode=require&application_name=svc-orders"
)

JSONB 的三套写法对照（同一张表）：

from sqlalchemy.dialects.postgresql import JSONB

# SQLAlchemy 模型
class Event(Base):
    __tablename__ = "events"
    id = Column(Integer, primary_key=True)
    payload = Column(JSONB)

# psycopg2 裸驱动
cur.execute("INSERT INTO events (payload) VALUES (%s)", (Json(data),))

# psycopg3 裸驱动：dict 直接传
cur.execute("INSERT INTO events (payload) VALUES (%s)", (data,))

行工厂与 typing：psycopg3 用 row_factory=dict_row 让 cursor 返回 dict，配合 Annotated 类型提示即可获得接近 ORM 的可读性，却保持裸 SQL 的性能与可控性——轻量项目的甜点区。

数据类型映射：从 PG 类型到 Python 对象

三套驱动对 PostgreSQL 类型的默认映射并不一致，金额、JSON、数组三类高频类型尤其要过一遍：

PostgreSQL 类型	psycopg2	psycopg3	asyncpg
integer / bigint	int	int	int
numeric	Decimal	Decimal	Decimal
float8	float	float	float
bool	bool	bool	bool
timestamptz	datetime（带 tzinfo）	datetime	datetime
json / jsonb	字符串（写侧需 Json 包装）	dict/list 自动双向	字符串（需 codec）
text[]	Python list	Python list	Python list
uuid	uuid.UUID（2.8+ 自动）	uuid.UUID	uuid.UUID
bytea	memoryview / bytes	bytes	bytes

工程上最常见的三个坑：

1. psycopg2 的 jsonb 是「半自动」：读出来默认是字符串，写进去要 psycopg2.extras.Json(data) 包装；想双向自动转换可以 register_default_jsonb，或者干脆用 psycopg3——后者开箱即得 dict ↔ jsonb。

2. asyncpg 的 jsonb 需要显式 codec，通常放在连接初始化回调里统一注册：

import json

async def init_conn(conn):
    await conn.set_type_codec(
        "jsonb",
        encoder=json.dumps,
        decoder=json.loads,
        schema="pg_catalog",
    )

pool = await asyncpg.create_pool(dsn, init=init_conn)

3. numeric 返回 Decimal 是精度保护而不是麻烦：金额计算留在 Decimal 域（Decimal("0.1") 永远不会变成 0.30000000000000004 问题的受害者），序列化给前端时用 str(Decimal) 保持字面精度，转 float 前必须三思。

数组的方向是「自动但扁平」：text[] 直接往返 Python list，多维数组与空数组边界（ARRAY[]::text[] 需要显式类型标注）偶有意外，复杂嵌套场景建议改用 jsonb 承载。完整映射表以各驱动官方文档为准，本节清单覆盖业务字段的绝大多数日常判断。

错误处理与异常类对照

三个库把 SQLSTATE 表达成不同的 Python 类型：

库	唯一冲突	序列化失败	通用入口
psycopg2	psycopg2.errors.UniqueViolation	psycopg2.errors.SerializationFailure	e.pgcode（SQLSTATE 字符串）
psycopg3	psycopg.errors.UniqueViolation	psycopg.errors.SerializationFailure	e.sqlstate
asyncpg	asyncpg.UniqueViolationError	asyncpg.SerializationError	e.sqlstate

from psycopg import errors

try:
    conn.execute("INSERT INTO users (email) VALUES (%s)", (email,))
except errors.UniqueViolation:
    raise ValueError("email already registered") from None
except errors.LockNotAvailable:
    ...          # 54P01：锁等待超时

psycopg3 的 psycopg.errors 按官方错误码层级组织（errors.lookup("23505") 可从 SQLSTATE 反查异常类）。混合库团队把三种异常收敛到统一的领域异常层——与 Go 的 *pgconn.PgError、Java 的 ServerErrorMessage 是同一模式在三种语言里的表达。

实战示例

示例一：psycopg2 标准业务事务

def transfer(conn, from_id: int, to_id: int, amount: int):
    try:
        with conn:
            with conn.cursor() as cur:
                cur.execute(
                    "UPDATE accounts SET balance = balance - %s "
                    "WHERE id = %s AND balance >= %s",
                    (amount, from_id, amount),
                )
                if cur.rowcount != 1:
                    raise ValueError("insufficient balance")
                cur.execute(
                    "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                    (amount, to_id),
                )
    except Exception:
        raise          # with conn 已回滚，向上抛出由调用方决定重试策略

示例二：execute_values 十万行插入

from psycopg2.extras import execute_values

rows = [(i, f"item-{i}", round(i * 0.1, 2)) for i in range(100_000)]
with conn.cursor() as cur:
    cur.execute("TRUNCATE stg_items")
    execute_values(
        cur,
        "INSERT INTO stg_items (id, name, price) VALUES %s",
        rows, page_size=2000,
    )
    conn.commit()

示例三：psycopg3 COPY 导入并带校验

import psycopg
from psycopg.rows import dict_row

def import_csv(dsn: str, csv_path: str, target: str):
    with psycopg.connect(dsn) as conn, conn.cursor(row_factory=dict_row) as cur:
        cur.execute(f"TRUNCATE {target}")
        with cur.copy(
            f"COPY {target} (event_id, user_id, kind, event_time) "
            "FROM STDIN WITH (FORMAT csv, HEADER true)"
        ) as cp:
            with open(csv_path, "r") as f:
                next(f)                    # 跳过表头（HEADER true 时也可不跳）
                for line in f:
                    cp.write(line)
        cur.execute(f"SELECT count(*) AS n FROM {target}")
        print("imported:", cur.fetchone()["n"])

示例四：ETL 小脚本（staging → 清洗 → 落表）

import psycopg
from psycopg.rows import dict_row

DSN = "host=localhost dbname=warehouse user=etl"

def run(csv_path: str):
    with psycopg.connect(DSN) as conn, conn.cursor(row_factory=dict_row) as cur:
        # 1) 装载到 staging
        cur.execute(
            """
            CREATE UNLOGGED TABLE IF NOT EXISTS stg_sales (
                order_id text, user_id text, amount text,
                currency text, ts text
            )
            """
        )
        cur.execute("TRUNCATE stg_sales")
        with cur.copy("COPY stg_sales FROM STDIN WITH (FORMAT csv, HEADER true)") as cp:
            for line in open(csv_path):
                cp.write(line)

        # 2) SQL 清洗 + 类型收敛 + 去重落表
        cur.execute(
            """
            INSERT INTO sales (order_id, user_id, amount, currency, sold_at)
            SELECT order_id::bigint,
                   user_id::bigint,
                   amount::numeric(12,2),
                   currency,
                   ts::timestamptz
            FROM (
                SELECT *, row_number() OVER (
                    PARTITION BY order_id ORDER BY ts DESC
                ) AS rn
                FROM stg_sales
                WHERE amount ~ '^-?[0-9]+(\\.[0-9]+)?$'
            ) clean
            WHERE rn = 1
            ON CONFLICT (order_id) DO NOTHING
            """
        )
        print("loaded:", cur.rowcount)
    # 退出 with 自动提交整个 ETL

示例五：FastAPI + asyncpg 池的异步查询

import asyncpg
from fastapi import FastAPI

app = FastAPI()
POOL = None

@app.on_event("startup")
async def startup():
    global POOL
    POOL = await asyncpg.create_pool(
        "postgresql://app_ro:secret@db1:5432/appdb", min_size=2, max_size=10
    )

@app.get("/users/{uid}")
async def get_user(uid: int):
    async with POOL.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, name, created_at FROM users WHERE id = $1", uid
        )
        return dict(row) if row else {"error": "not found"}

注意事项

%s 是占位符不是格式化。 psycopg 系的参数永远是 execute(sql, (a, b)) 形式；sql % vals、f-string、.format() 拼接用户输入都是注入事故。IN 查询用 = ANY(%s) 传数组最省心。

psycopg2 的 executemany 不是批量。 它是逐条循环；十万行级写入用 execute_values() 或 COPY，性能差一个数量级以上。

只读查询也要 commit。 psycopg2 默认事务模式下 SELECT 同样开启事务，不提交会长时间挂起只读事务；要么 autocommit，要么查询后 commit。这是「vacuum 不推进、复制延迟」类疑难杂症的常见根因。

fetchall 会把结果全部装进内存。 大结果集用命名游标（psycopg2）分块迭代、psycopg3 的 cursor 流式特性、或直接在 SQL 层分页。

psycopg3 的二进制与扩展协议默认开启。 绝大多数场景是收益；个别语句需要文本协议调试时按官方文档显式切换，不要凭感觉关闭。

asyncpg 的 $1 占位符与 psycopg 的 %s 不通用。 团队混用两个库时，把「驱动占位符风格」写进代码规范，避免复制粘贴事故。

JSONB 适配器版本分明。 psycopg2 要 Json() 包装，psycopg3 直接传 dict；混用库里最容易出现「写进去是字符串不是 jsonb」的静默错误，入库后用 SELECT jsonb_typeof(col) 抽查。

连接池参数要对着 PostgreSQL 配置算。 应用池 max_size × 实例数 ≤ PostgreSQL max_connections 余量；再叠加 PgBouncer 时应用池相应收小，两层池的总和才是数据库承受的真实并发。

自动预处理的阈值不用调。 psycopg3 的 prepare_threshold 默认值已是合理分档；把它归零「优化」反而让低频语句也吃服务端语句缓存的成本，还会放大与 PgBouncer transaction 模式的兼容面。

最佳实践

1. 新项目默认 psycopg3，存量库随升级迁移。 类型映射原生、copy/pipeline/池一等公民，迁移成本主要在异常类与 DSN 风格。
2. 连接参数显式声明 connect_timeout、sslmode、application_name。 三者分别对应「网络故障快速失败」「传输加密」「连接可溯源」，缺一不可。
3. 批量写入分档：executemany（小）/ execute_values（中）/ COPY（大）。 十万行是 COPY 的分水岭；psycopg3 的 executemany 已做多值优化，但仍不及 COPY。
4. ETL 一律 staging 表 + SQL 清洗。 类型收敛、去重、约束校验都在跳板表完成，坏数据到不了正式表。
5. 事务边界用 with conn / with conn.transaction() 表达。 让提交与回滚由语言结构保证，而不是散落的手动 commit。
6. asyncio 服务用 asyncpg 或 psycopg3 AsyncPool。 事件循环里跑同步 psycopg2 是把并发优势整段放弃。
7. 连接池必配健康检查与超时。 psycopg_pool 自带；psycopg2 池过朴素，配合 PgBouncer 使用更稳。
8. 金额用 numeric + Decimal，绝不过 float。 Python 的 Decimal 与 PostgreSQL 的 numeric 是精度安全的天作之合。
9. 大结果集用服务端游标流式处理。 itersize（psycopg2）与 chunked copy（psycopg3）让内存曲线保持水平。
10. SQLAlchemy 项目用方言 URL 显式选驱动。 +psycopg/+asyncpg 写进配置，升级驱动只改一行。
11. 用 pg_stat_activity 的 application_name 做连接溯源。 每个服务一个名字，慢连接、悬挂事务一眼定位。
12. 高延迟链路启用 pipeline 模式。 psycopg3 的 pipeline 在跨机房、云上跨可用区场景收益显著，一批小查询合批发送。

下一节

继续学习：Java 连接 PostgreSQL

参考链接

• psycopg2 官方文档
• psycopg2 extras（execute_values / Json / RealDictCursor）
• psycopg3 官方文档
• psycopg3：COPY 指南
• psycopg_pool 连接池文档
• asyncpg 官方文档
• SQLAlchemy PostgreSQL 方言
• PostgreSQL 官方文档：COPY
• PostgreSQL 官方文档：libpq 连接参数