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

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

SELECT data->>'status' FROM orders;

SELECT * FROM events WHERE data ? 'error_code';

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

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

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

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

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]}")

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

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'])

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

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

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

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

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 '{}'
);

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

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

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