如何在Flask-CQLAlchemy中定义带验证的自定义列类型?
Flask-CQLAlchemy中实现带时区校验的DateTime自定义类型
我正在用flask-cqlalchemy在Flask应用中连接Cassandra,当前定义Cassandra模型的方式如下:
from flask_cqlalchemy import CQLAlchemy # app 是Flask应用实例 cassandra_db = CQLAlchemy(app) class CassandraBase(cassandra_db.Model): column1 = cassandra_db.columns.DateTime(default=lambda: datetime.now(tz=pytz.UTC))
我知道可以通过继承cassandra_db.UserType定义自定义用户类型,文档中的示例如下:
class Address(cassandra_db.UserType): street = cassandra_db.columns.Text() zipcode = cassandra_db.columns.Integer()
但我想要定义一个DateTimeTz自定义类型,强制要求传入带时区信息的时间,且不需要在类型中指定列名,示例结构如下:
class DateTimeTz: # 如果用户传入无时区的datetime,抛出`ValidationError` # 不需要在此指定列名
期望能在所有Cassandra模型中这样使用:
class CassandraBase(cassandra_db.Model): column1 = DateTimeTz(default=lambda: datetime.now(tz=pytz.UTC))
目前我在模型类中添加了验证逻辑:
class CassandraBase(cassandra_db.Model): column1 = cassandra_db.columns.DateTime(default=lambda: datetime.now(tz=pytz.UTC)) def validate(self): # 重写基类`Model`的`validate`方法 super().validate() if self.column1 and not self.column1.tzinfo: raise ValidationError('column1未传入时区信息')
但这个方案存在问题:每个需要校验时区的Cassandra模型都要重写validate方法,为所有相关DateTime列添加判断。我查阅了文档和网络资料,只找到这种重写validate的方式,而Flask-SQLAlchemy中可以用TypeDecorator基类实现类似功能:
from sqlalchemy.types import TypeDecorator class DateTimeTZ(TypeDecorator): impl = DateTime cache_ok = True def process_bind_param(self, value, dialect): if value is not None and not value.tzinfo: raise ValueError("必须传入带时区的时间") return value
请问在Flask-CQLAlchemy中操作Cassandra,有没有更好的实现方式?
内容的提问来源于stack exchange,提问作者Ashutosh
相关产品推荐
相关产品推荐

