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

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

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

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

一、postgresql 中的 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

1.1 json vs jsonb：关键区别

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

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

1.2 常用操作符与函数

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"');

二、python 驱动选择与配置

2.1 主流驱动对比

驱动	异步支持	json 自动转换	成熟度	适用场景
`psycopg2`	否	需手动注册适配器	高	同步应用、django、flask
`psycopg2-binary`	否	同上	高	快速原型、无需编译
`asyncpg`	是	自动转换 dict ↔ jsonb	高	异步框架（fastapi, aiohttp）
`sqlalchemy` + `psycopg2`	否	通过 `json` / `jsonb` 类型自动处理	极高	orm 场景

2.2 postgresql 的 jsonb 与 python 的结合注意点

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

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

但也要谨记：

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

三、使用 psycopg2 处理 json

3.1 基础连接与自动转换

默认情况下，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"
)

3.2 插入 json 数据

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，否则会被当作字符串插入。

3.3 查询并自动反序列化

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。

3.4 复杂查询示例

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 处理 json（异步场景）

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

4.1 基础用法

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())

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

4.2 复杂查询

# 使用路径操作符
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 orm 处理 json

sqlalchemy 通过 json 和 jsonb 类型提供 orm 支持。

5.1 模型定义

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)

5.2 crud 操作

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()

5.3 使用专用函数（sqlalchemy 1.4+）

from sqlalchemy.dialects.postgresql import jsonb

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

六、高级技巧与最佳实践

6.1 动态构建 json 查询条件

避免拼接 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")

6.2 更新 json 字段

# 使用 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 表示：若路径不存在则创建。

6.3 索引优化

对高频查询字段建立 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 创建。

七、性能考量

7.1 存储效率

jsonb 比 json 节省 10%~30% 空间
避免在 json 中存储大文本（如 base64 图片），应存 url

7.2 查询性能

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

7.3 批量操作

使用 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]
)

八、典型应用场景

8.1 用户配置存储

create table user_settings (
    user_id int primary key,
    preferences jsonb not null default '{}'
);

优势：无需 alter table 即可新增配置项
查询：select preferences->'theme' from user_settings

8.2 日志与事件追踪

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

8.3 动态表单/问卷

表单结构存为 json，回答存为另一 json
避免 eav（entity-attribute-value）反模式

九、常见陷阱与解决方案

9.1 时区与日期处理

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;

9.2 精度丢失（浮点数）

postgresql 的 jsonb 使用 ieee 754 双精度，与 python 一致，一般无问题。
但若需高精度（如金融），应存为字符串或使用 numeric 字段。

9.3 键名大小写敏感

json 对象键区分大小写：

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

建议统一使用小写命名。

以上就是使用python与postgresql的json数据进行交互的详细内容，更多关于python与postgresql json数据交互的资料请关注代码网其它相关文章！

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

2026年02月07日 • Python •我要评论