前言

SQLAlchemy 版本：1.2

Mysql 这种数据库不支持 json 对象的存储，但是开发过程中又确实有这种需求。一种思路是使用字符串来存储该对象，入库时序列化，出库时反序列化。在业务代码中维护数据转换又很大的思维负担，还容易出错，因此这个功能的实现自然而然落在了 ORM 里。

自定义类型

SQLAlchemy 支持自定义处理函数，继承 TypeDecorator 即可很方便得到用户自定义类型。代码如下：

import json
from sqlalchemy.types import TypeDecorator, String
class JSONEncodedDict(TypeDecorator):
    """Represents an immutable structure as a json-encoded string.

    Usage::
        JSONEncodedDict(255)
    """

    impl = String

    def process_bind_param(self, value, dialect):
      """Receive a bound parameter value to be converted.
      """
        if value is not None:
            value = json.dumps(value)

        return value
      
    def process_result_value(self, value, dialect):
      """Receive a result-row column value to be converted.
      """
        if value is not None:
            value = json.loads(value)
        return value

这段代码扩展了原生的 String 类型，在处理 bind param 时将其序列化，在从数据库取出结果后反序列化。还有其他的可定义的步骤，如 process_literal_param，定义了如何处理一个 literal param。但是这里我用不到，就不定义了。

定义 Model

这里我踩到一个坑，结合 Model 定义来趟坑。

# 定义数据库
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:123456@localhost:3306/public?charset=utf8mb4', echo=True)

# 定义 Model
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy import Column, Integer, String, text
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=True)
    fullname = Column(String(100), server_default=text("""'this is text default'"""))
    password = Column(String(100), server_default="'this is direct default'")
    wrong_user_ids = Column(JSONEncodedDict(255), default='["this is default"]')
    right_user_ids = Column(JSONEncodedDict(255), default=["this is default"])
Base.metadata.create_all(engine)

以下为打印出来的建表语句，可以看出，default 参数不影响建表语句，server_default 会影响，text() 函数将后面的 server_default 直接放在生成 SQL 对应位置，不使用该函数则会对 server_default 进行对应的转义。

CREATE TABLE users (
	id INTEGER NOT NULL AUTO_INCREMENT, 
	name VARCHAR(100), 
	fullname VARCHAR(100) DEFAULT 'this is text default', 
	password VARCHAR(100) DEFAULT '''this is direct default''', 
	right_user_ids VARCHAR(255), 
	wrong_user_ids VARCHAR(255), 
	PRIMARY KEY (id)
)

下面生成一个数据

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
session.add(User(fullname='Hieast', password='123456'))
session.commit()
obj = session.query(User).first()
print(obj.right_user_ids)
print(obj.wrong_user_ids)

print 的输出是

1 2	([u'this is default'], <type 'list'>) (u'["this is default"]', <type 'unicode'>)

可以看到，default 在写入数据库的时候，是作为 bind_param 传入的，因此要经过 process_bind_param 的处理。如果我们定义的 process_bind_param 期望的输入是一个数组或字典，那么 default 也要赋值一个数组或字典，而不是已经序列化之后的字符串。