SQLAlchemy自定义数据类型

有时，我们想把一个dict对象存到MySQL表里面，由于MySQL没有定义dict之类的数据类型，一种做法是，每当往表里写入数据时，先将dict对象转换成字符串形式，每当从表里读取数据时，将读出来的字符串转换成dict对象。如果应用开发者自己显式地处理数据转换，显得麻烦并且较易出错。

SQLAlchemy允许自定义数据类型来扩展已有数据类型。为了定义新的数据类型，需要继承TypeDecorator类，并且定义bind-parameter和result-processing行为。bind-parameter行为决定在写入数据库时如何转换数据，result-processing行为决定在读取数据库时如何转换数据。

了解MongoDB的人一定知道ObjectId，它是MongoDB文档的Id。假设我们需要将ObjectId存到MySQL里面，而MySQL没有类似ObjectId的数据类型，我们定义一个新的数据类型。由于ObjectId有其二进制和字符串形式的表示，我们将要自定义的数据类型可以在已有MySQL数据类型BINARY或者VARCHAR的基础上扩展。下面以BINARY作为示例。

import sqlalchemy.types as types
from bson import ObjectId
 
class MysqlObjectId(types.TypeDecorator):
 """ custom type
  Usage:
   MysqlObjectId(12)
  ObjectId's binary representation has 12 bytes.
 """
  
 impl = types.BINARY
  
 def process_bind_param(self, value, dialect):
  """define bind-parameter behavior
  """
  if value is not None:
   try:
    # when value is not None,
    # ObjectId's __init__ method will validate if value is illegal
    # return the binary representation) of ObjectId
    return ObjectId(value).binary
   except:
    pass
  return None
   
 def process_result_value(self, value, dialect):
  """define result-processing behavior
  """
  if value is not None:
   try:
    return ObjectId(value)
   except:
    pass
  return None

上面定义了MysqlObjectId类型，下面的代码示例了如何使用该自定义类型。

import sqlalchemy as SA 
from sqlalchemy.ext.declarative import declarative_base
 
Base =  declarative_base()
   
class TestTable(Base):
  
 __tablename__ = "test_table"
  
 # column definitions
 id = SA.Column(SA.Integer, primary_key=True, autoincrement=True)
 object_id = SA.Column(MysqlObjectId(12), nullable=True)

测试代码就不写了。