个人技术分享

在Java的ORM框架Mybatis和JPA中，都可以自定义数据库类型转化为实体类属性的类型转化器，Mybatis中是TypeHandler，JPA中是AttributeConverter，这种灵活定制的方式在某些场景下很有用，譬如枚举在数据库中村的是int，数据库中的时间戳需要转化为Datetime等。
最近在使用python的orm框架sqlalchemy，学习了自定义类型转换器。

定义一个类，继承自sqlalchemy.types.TypeDecorator，配置impl和cache_ok两个属性，再实现两个方法即可。

• TypeDecorator，sqlalchemy内置，继承该类方便自定义类型。
• impl，类属性，由数据库类型决定，无需考虑python类型。
• cache_ok，对于同一个值，相互转化后的值是否一样，一样则表示可缓存。
• process_bind_param(self, value, dialect: Dialect)，将实体属性转化为数据库值。
• process_result_value(self, value, dialect: Dialect)，数据库值转化为实体属性。

import datetime
import enum
import logging
from typing import Optional, Dict, Type
from urllib.parse import quote_plus

from sqlalchemy import create_engine, Integer, Text, Dialect
from sqlalchemy import types
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

import settings

# 定义枚举
class Tag(enum.Enum):
    other = 0
    passenger_transport = 1

# 数据库中存储时间格式字符串，查询后转python date类型
class StrDateType(types.TypeDecorator):
    # 由数据库类型决定，impl无需考虑python类型
    impl = types.CHAR
    # 对于同一个值，相互转化后的值是否一样，一样则表示可缓存
    cache_ok = True

    def __init__(self, *args, **kwargs):
        self._date_format = kwargs.pop('format', '%Y-%m-%d')
        super().__init__(*args, **kwargs)

    def process_bind_param(self, value: Optional[datetime.date], dialect: Dialect) -> Optional[str]:
        if value is None:
            return None
        return value.strftime(self._date_format)

    def process_result_value(
            self, value: Optional[str], dialect: Dialect
    ) -> Optional[datetime.date]:
        if not value:
            return None
        return datetime.datetime.strptime(value, self._date_format).date()

# 通用枚举类型转化
class GenericDbEnum(types.TypeDecorator):
    impl = Integer
    cache_ok = True

    def __init__(self, enum_cls: Type[enum.Enum], *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._map: Dict[int, enum_cls] = {e.value: e for e in enum_cls.__members__.values()}

    def process_bind_param(self, value: Optional[enum.Enum], dialect: Dialect) -> Optional[int]:
        return None if value is None else value.value

    def process_result_value(
            self, value: Optional[int], dialect: Dialect
    ) -> Optional[enum.Enum]:
        return None if value is None else self._map[value]


class Base(DeclarativeBase):
    pass


class DDSInfo(Base):
    __tablename__ = "dds_info_gt"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, nullable=False)
    date: Mapped[datetime.date] = mapped_column("rq", StrDateType(10, format='%Y%m%d'), nullable=False)
    flight: Mapped[int] = mapped_column("bc", Integer, nullable=False)
    text: Mapped[str] = mapped_column("aq6", Text, nullable=False)
    tag: Mapped[Tag] = mapped_column("bz", GenericDbEnum(Tag), nullable=True)

engine = create_engine(
    'mysql+pymysql://root:{password}@192.168.8.3/GRPT'.format(password=quote_plus('abc@123')),
    echo=True, pool_recycle=3600)
# 不存在则建表
Base.metadata.create_all(engine)