Python & SQLALchemy 学习笔记_JSON 栏位操作

这边纪录了一下 JSON 栏位的操作方式，
有任何错误或是更好的写法，欢迎留言讨论喔~

一、建立资料以及表结构

(一)、建立表结构

建立表结构如果有不清楚的地方可以看这篇文章

main.py
注: 突然改用 mysql 的原因是，在使用 sqlite 写入 JSON 时，发现中文写不进去
注: 已解决，参考这个连结

# -*- coding: utf-8 -*-
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import Integer, JSON
from sqlalchemy.orm import sessionmaker

Base = declarative_base()
engine_url = "mysql+pymysql://root:[email protected]:3306/test"
engine = create_engine(engine_url, echo=True)


class Test(Base):
    __tablename__ = "test"
    id = Column(Integer, primary_key=True, autoincrement=True)
    records = Column(JSON)


def create_table():
    Base.metadata.create_all(engine)


def drop_table():
    Base.metadata.drop_all(engine)


def create_session():
    Session = sessionmaker(bind=engine)
    session = Session()

    return session


if __name__ == '__main__':
    drop_table()
    create_table()

(二)、建立测试资料

对於新增资料有不懂的地方可以参考这篇文章

test.py

import json
from data_struct import Test, create_session
from pprint import pprint

with open("./test_data.json", "rb") as f:
    data = json.load(f)
    pprint(data[0])

    session = create_session()
    for row in data[0]["result"]["records"]:
        data_obj = {"records": row}
        session.add(Test(**data_obj))
        session.commit()

    session.close()

二、利用 SQL 语法查询

语法结构:
SELECT * FROM test WHERE JSON_EXTRACT(<栏位名称>, "$.<key>")=<条件>;
简单范例:
SELECT * FROM test WHERE JSON_EXTRACT(records, "$.Seq")=1;
中文 key:
如果 key 值为中文，则必须 key 部分的语法改为 '$."<中文 key>"'
简单范例:
SELECT * FROM test WHERE JSON_EXTRACT(records, '$."性别"')="男";

三、利用 sqlalchemy 查询

语法结构:
同样支援 all() 和 first()
session.query(<表结构>).filter(<表结构>.<栏位>[key] == <条件>).all()
简单范例

# -*- coding: utf-8 -*-
from main import Test, create_session
from pprint import pprint

session = create_session()
res = session.query(Test).filter(Test.records["性别"] == "女").all()

for row in res:
    print(row.id, end=" ")
    print(row.records["Seq"], end=" ")
    print(row.records["性别"], end=" ")
    print(row.records["级别"])

output:

C:\Users\bear\Desktop\sqlalchemy_test\venv\Scripts\python.exe C:/Users/bear/Desktop/sqlalchemy_test/test.py
11 11 女 初级
12 12 女 中级
13 13 女 中高级
14 14 女 高级
15 15 女 优级
16 16 女 初级
17 17 女 中级
18 18 女 中高级
19 19 女 高级
20 20 女 优级

Process finished with exit code 0

三、利用 sqlalchemy 更新

语法结构:

from sqlalchemy import func

session.query(<表结构>).filter(<条件>).update({<表结构>.<栏位>: func.json_set(<表结构>.<栏位>, "$.<key>", <要更改的内容>))})

简单范例:

session = create_session()

session.query(Test).filter_by(id=1).update({Test.records: func.json_set(Test.records, "$.Seq", "测试测试测试")})

session.commit()
session.close()

注: 最後要记得 commit 资料库中的资料才会更改