使用Python与PostgreSQL的JSON数据进行交互

目录

• 一、PostgreSQL 中的 JSON 类型基础
• 1.1 JSON vs JSONB：关键区别
• 1.2 常用操作符与函数
• 二、Python 驱动选择与配置
• 2.1 主流驱动对比
• 2.2 PostgreSQL 的 JSONB 与 Python 的结合注意点
• 三、使用 psycopg2 处理 JSON
• 3.1 基础连接与自动转换
• 3.2 插入 JSON 数据
• 3.3 查询并自动反序列化
• 3.4 复杂查询示例
• 四、使用 asyncpg 处理 JSON（异步场景）
• 4.1 基础用法
• 4.2 复杂查询
• 五、使用 SQLAlchemy ORM 处理 JSON
• 5.1 模型定义
• 5.2 CRUD 操作
• 5.3 使用专用函数（SQLAlchemy 1.4+）
• 六、高级技巧与最佳实践
• 6.1 动态构建 JSON 查询条件
• 6.2 更新 JSON 字段
• 6.3 索引优化
• 七、性能考量
• 7.1 存储效率
• 7.2 查询性能
• 7.3 批量操作
• 八、典型应用场景
• 8.1 用户配置存储
• 8.2 日志与事件追踪
• 8.3 动态表单/问卷
• 九、常见陷阱与解决方案
• 9.1 时区与日期处理
• 9.2 精度丢失（浮点数）
• 9.3 键名大小写敏感

PostgreSQL 自 9.2 版本起原生支持 JSON 数据类型，并在后续版本中不断增强其功能，现已提供 JSON 和 JSONB 两种类型、丰富的操作符、索引支持及函数体系。与此同时，Python 作为数据处理的主流语言，与 PostgreSQL 的结合日益紧密。

本文将系统讲解如何在 Python 中高效、安全、可维护地与 PostgreSQL 的 JSON 数据交互，涵盖：

• JSON 与 JSONB 的区别与选型
• 使用 psycopg2 和 asyncpg 等主流驱动
• 自动序列化/反序列化 Python 字典与 JSON
• 复杂查询：路径提取、条件过滤、更新操作
• 性能优化与索引策略
• 实战案例：配置存储、日志分析、动态表单等

参考

• PostgreSQL JSON 函数文档：https://www.postgresql.org/docs/current/functions-json.html
• psycopg2 JSON 支持：https://www.psycopg.org/docs/extras.html#json-adaptation
• asyncpg 类型映射：https://magicstack.github.io/asyncpg/current/usage.html#type-conversion

特性	JSON	JSONB
存储格式	文本（保留原始格式）	二进制（解析后存储）
是否去重	否（保留重复键）	是（仅保留最后一个键值）
是否保留顺序	是	否（对象键无序）
索引支持	不支持（需表达式索引）	支持 GIN、GiST 索引
查询性能	较慢（每次需解析）	快（已解析为内部结构）
存储空间	较大	较小（无空白、无重复）

推荐：除非必须保留原始 JSON 格式（如审计日志），否则一律使用 JSONB。

1、路径提取

->：返回 JSON 对象（仍为 JSONB）

SELECT data->'user'->'name' FROM logs;

->>：返回文本（TEXT）

SELECT data->>'status' FROM orders;

2、条件查询

检查键是否存在：

SELECT * FROM events WHERE data ? 'error_code';

检查嵌套路径：

SELECT * FROM configs WHERE data @> '{"feature": {"enabled": true}}';

3、更新操作

设置字段（PostgreSQL 9.5+）：

UPDATE users SET profile = jsonb_set(profile, '{settings,theme}', '"dark"');

驱动	异步支持	JSON 自动转换	成熟度	适用场景
psycopg2	否	需手动注册适配器	高	同步应用、Django、Flask
psycopg2-binary	否	同上	高	快速原型、无需编译
asyncpg	是	自动转换 dict ↔ JSONB	高	异步框架（FastAPI, aiohttp）
SQLAlchemy + psycopg2	否	通过 JSON / JSONB 类型自动处理	极高	ORM 场景

PostgreSQL 的 JSONB 与 Python 的结合，为处理半结构化数据提供了强大而灵活的方案。通过合理选择驱动、配置自动转换、利用索引和操作符，可实现：

• 开发效率高：无需预定义 schema
• 查询能力强：支持复杂嵌套查询
• 性能可优化：GIN 索引、路径索引保障效率

但也要谨记：

• 不要滥用 JSON：结构化数据仍应使用关系模型
• 保持查询简单：避免过度嵌套导致维护困难
• 监控性能：定期分析执行计划

默认情况下，psycopg2 将 JSONB 列返回为字符串。需注册适配器实现自动转换：

import json
import psycopg2
from psycopg2.extras import Json

# 注册自动转换：DB → Python
def _json_decode(data, cur):
    if data is None:
        return None
    return json.loads(data)

# 注册适配器
psycopg2.extensions.register_type(
    psycopg2.extensions.new_type(
        (3802,), "JSONB", _json_decode  # 3802 是 JSONB 的 OID
    )
)

# 连接数据库
conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="user",
    password="pass"
)

data = {
    "user_id": 123,
    "action": "login",
    "metadata": {
        "ip": "192.168.1.1",
        "device": "mobile"
    }
}

cur = conn.cursor()
cur.execute(
    "INSERT INTO events (event_data) VALUES (%s)",
    (Json(data),)  # 使用 Json 包装器
)
conn.commit()

注意：必须使用 psycopg2.extras.Json，否则会被当作字符串插入。

cur.execute("SELECT id, event_data FROM events WHERE id = %s", (1,))
row = cur.fetchone()
print(type(row[1]))  # <class 'dict'>
print(row[1]['metadata']['ip'])  # 192.168.1.1

得益于前面的适配器注册，event_data 自动转为 dict。

1、提取嵌套字段

cur.execute("""
    SELECT 
        id,
        event_data->>'action' AS action,
        event_data->'metadata'->>'ip' AS ip
    FROM events
    WHERE (event_data->'metadata'->>'device') = %s
""", ("mobile",))

for row in cur:
    print(f"Action: {row[1]}, IP: {row[2]}")

2、条件过滤（使用 @>）

# 查找包含特定子结构的记录
filter_condition = {"metadata": {"device": "mobile"}}
cur.execute(
    "SELECT * FROM events WHERE event_data @> %s",
    (Json(filter_condition),)
)

asyncpg 对 JSONB 支持更友好，默认自动转换。

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect(
        host='localhost',
        database='mydb',
        user='user',
        password='pass'
    )

    # 插入：直接传 dict
    data = {"user_id": 123, "tags": ["a", "b"]}
    await conn.execute(
        "INSERT INTO items (payload) VALUES ($1)",
        data  # asyncpg 自动序列化为 JSONB
    )

    # 查询：自动反序列化为 dict/list
    row = await conn.fetchrow("SELECT payload FROM items LIMIT 1")
    print(type(row['payload']))  # <class 'dict'>
    print(row['payload']['tags'])  # ['a', 'b']

    await conn.close()

asyncio.run(main())

优势：无需额外配置，开箱即用。

# 使用路径操作符
rows = await conn.fetch("""
    SELECT 
        id,
        payload->'user'->>'name' AS name
    FROM profiles
    WHERE payload @> $1
""", {"settings": {"visible": True}})

for r in rows:
    print(r['name'])

SQLAlchemy 通过 JSON 和 JSONB 类型提供 ORM 支持。

from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class Event(Base):
    __tablename__ = 'events'
    id = Column(Integer, primary_key=True)
    data = Column(JSONB)  # 使用 JSONB

engine = create_engine('postgresql://user:pass@localhost/mydb')
Session = sessionmaker(bind=engine)

session = Session()

# 插入
event = Event(data={
    "type": "click",
    "element": {"id": "btn1", "class": "primary"}
})
session.add(event)
session.commit()

# 查询（自动转为 dict）
e = session.query(Event).first()
print(e.data['element']['id'])  # btn1

# 条件查询（使用 .op() 调用操作符）
from sqlalchemy import text
results = session.query(Event).filter(
    Event.data.op('@>')({'type': 'click'})
).all()

from sqlalchemy.dialects.postgresql import JSONB

# 提取字段
session.query(
    Event.data['user']['name'].astext.label('username')
).all()

避免拼接 SQL，使用参数化：

def find_by_metadata(conn, **kwargs):
    # 构建嵌套 dict
    condition = {"metadata": kwargs}
    cur = conn.cursor()
    cur.execute(
        "SELECT * FROM events WHERE event_data @> %s",
        (Json(condition),)
    )
    return cur.fetchall()

# 调用
results = find_by_metadata(conn, device="mobile", os="iOS")

# 使用 jsonb_set 更新嵌套字段
cur.execute("""
    UPDATE users 
    SET profile = jsonb_set(profile, %s, %s, true)
    WHERE id = %s
""", (
    ['settings', 'notifications'],  # 路径（数组）
    Json({"email": True, "push": False}),  # 新值
    user_id
))

第四个参数 true 表示：若路径不存在则创建。

对高频查询字段建立 GIN 索引：

-- 全文索引（适用于任意键查询）
CREATE INDEX idx_event_data ON events USING GIN (event_data);

-- 特定路径索引（更高效）
CREATE INDEX idx_event_action ON events ((event_data->>'action'));

在 Python 中可通过 Alembic 或原生 SQL 创建。

• JSONB 比 JSON 节省 10%~30% 空间
• 避免在 JSON 中存储大文本（如 base64 图片），应存 URL

• 避免在 WHERE 中对 JSON 字段使用函数（如 length(data->>'name')），会导致索引失效
• 优先使用 @>, ?, ->> 等支持索引的操作符

• 使用 execute_values（psycopg2）或 copy_records_to_table（asyncpg）提升批量插入性能
• 示例（psycopg2）：

from psycopg2.extras import execute_values
data_list = [{"id": i, "tags": ["x"]} for i in range(1000)]
execute_values(
    cur,
    "INSERT INTO items (payload) VALUES %s",
    [(Json(d),) for d in data_list]
)

CREATE TABLE user_settings (
    user_id INT PRIMARY KEY,
    preferences JSONB NOT NULL DEFAULT '{}'
);

• 优势：无需 ALTER TABLE 即可新增配置项
• 查询：SELECT preferences->'theme' FROM user_settings

• 存储非结构化日志，支持按任意字段过滤
• 结合 BRIN 索引按时间分区，GIN 索引按内容查询

• 表单结构存为 JSON，回答存为另一 JSON
• 避免 EAV（Entity-Attribute-Value）反模式

JSON 不支持日期类型，通常存为 ISO 字符串：

data = {"created_at": datetime.utcnow().isoformat()}

查询时用 to_timestamp() 转换：

SELECT to_timestamp(data->>'created_at', 'YYYY-MM-DD"T"HH24:MI:SS.US') 
FROM logs;

PostgreSQL 的 JSONB 使用 IEEE 754 双精度，与 Python 一致，一般无问题。
但若需高精度（如金融），应存为字符串或使用 NUMERIC 字段。

JSON 对象键区分大小写：

-- 以下不等价
data->'UserId'  vs  data->'userid'

建议统一使用小写命名。

以上就是使用Python与PostgreSQL的JSON数据进行交互的详细内容，更多关于Python与PostgreSQL JSON数据交互的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python连接数据库的3种方法对比(SQLiteMySQLPostgreSQL)
• Python使用psycopg2操作PostgreSQL数据库的完全指南
• Python执行PostgreSQL数据库的SQL脚本详解
• Python进行PostgreSQL数据库连接的详细使用指南
• python如何遍历postgresql所有的用户表