def as_mutable(cls, orig_sqltype):
"""Mark the value as nested mutable value.
What happens here
* We coerce the return value - the type value set on sqlalchemy.Column() to the underlying SQL typ
* We mark this type value with a marker attribute
* Then we set a global SQAlchemy mapper event handler
* When mapper is done setting up our model classes, it will call the event handler for all models
* We check if any of the models columns have our marked type value as the value
* If so we call ``associate_with_attribute`` for this model and column that sets up ``MutableBase._listen_on_attribute`` event handlers. These event handlers take care of taking the raw dict coming out from database and wrapping it to NestedMutableDict.
:param orig_sqltype: Usually websauna.system.model.column.JSONB instance
:return: Marked and coerced type value
"""
# Create an instance of this type and add a marker attribute,
# so we later find it.
# We cannot directly compare the result type values, as looks like
# the type value might be mangled by dialect specific implementations
# or lost somewhere. Never figured this out 100%.
sqltype = types.to_instance(orig_sqltype)
sqltype._column_value_id = id(sqltype)
def listen_for_type(mapper, class_):
for prop in mapper.column_attrs:
# The original implementation has SQLAlchemy type comparator.
# Here we need to be little more complex, because we define a type alias
# for generic JSONB implementation
if getattr(prop.columns[0].type, "_column_value_id", None) == sqltype._column_value_id:
cls.associate_with_attribute(getattr(class_, prop.key))
event.listen(mapper, 'mapper_configured', listen_for_type)
return sqltype
评论列表
文章目录
