How to detect changes in custom types in sqlalchemy

Question

I am working on custom types in sqlalchemy columns using TypeDecorator. I'm storing my data in JSONB inside Postgres DB, but in code I am serializing and deserializing it in a data class. But when I change any field of that data class, it didn't detect changes on that column. How can I achieve this in sqlalchemy.

dataclass :

@dataclass
class ConfigFlags(DataClassJsonMixin):
    abc: bool = True

custom type def :

class ConfigFlagType(types.TypeDecorator):

    """Type for config flags."""

    impl = JSONB

    def process_result_value(  # noqa: ANN201
        self,
        value: Optional[dict],
        dialect,  # noqa: ANN001
    ):
        if not isinstance(value, dict):
            msg = "Value must be a dictionary"
            raise ValueError(msg)  # noqa: TRY004

        return ConfigFlags.from_dict(value)

    def process_bind_param(  # noqa: ANN201
        self,
        value: Optional[ConfigFlags],
        dialect,  # noqa: ANN001
    ):
        if not isinstance(value, ConfigFlags):
            msg = "Value must be a ConfigFlags"
            raise ValueError(msg)  # noqa: TRY004
        return value.to_dict()

db model :

class ConvHistories(CBase):

    """conv_histories model."""

    __tablename__ = "test"

    id = Column(Integer, primary_key=True, autoincrement=True)
    config_flags: ConfigFlags = Column(
        ConfigFlagType,
    )

   def find_one(self, conv_id: int) -> "ConvHistories":
        return self.query.filter(  
            ConvHistories.id == conv_id,
        ).first()

res = ConvHistories(session=session).find_one(conv_id=3)

if res.config_flags:
    res.config_flags.abc = False


session.add(res)
session.commit()

But it didn't detect changes in config_flags column. How can I do this?

Answer 1

From the SqlAlchemy JSON docs (which also apply to JSONB):

The JSON type, when used with the SQLAlchemy ORM, does not detect in-place mutations to the structure. In order to detect these, the sqlalchemy.ext.mutable extension must be used, most typically using the MutableDict class. This extension will allow “in-place” changes to the datastructure to produce events which will be detected by the unit of work. See the example at HSTORE for a simple example involving a dictionary.

So, if you had a simple column, you'd wrap it like this:

config_flags = Column(MutableDict.as_mutable(JSONB))

In your case it's a bit more involved because you have a special wrapper class, but you still need to use the same mechanism.

More sample code + very good explanation can be found in this blog post Beware of JSON fields in SQLAlchemy

As simpler, but probably less efficient alternative to using the mutation tracking of MutableDict, you may also be able to, whenever you modify a config_flag, first make a deep copy and then add it back:

if res.config_flags:
   res.config_flags = copy.deepcopy(res.config_flags)
   res.config_flags.abc = False

   session.add(res)
   session.commit()