PostgreSQL Question posted in
The official documentation can be found here.

Postgresql – How to detect changes in custom types in sqlalchemy

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?

Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    After digging too much in sqlalchemy I came across with below solution.

    First, create a class similar to sqlalchemy.ext.mutable.MutableDict for objects as shown here.

    class MutableObject(Mutable):
    @classmethod
    def coerce(cls, key, value):
        return value

    def __getstate__(self):
        d = self.__dict__.copy()
        d.pop("_parents", None)
        return d

    def __setstate__(self, state):
        self.__dict__ = state

    def __setattr__(self, name, value) -> None:
        old_val = object.__getattribute__
        object.__setattr__(self, name, value)
        self.changed()

    In above code snippet, object marked changed when __setattr__ called.

    Now declare our data class and custom type as shown below

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

    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(
        MutableObject.as_mutable(ConfigFlagType),
    )

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

    This solution works for me.

    Note: One issue I am facing with this solution is in data classes initialization also take place using __setattr__ method. So when class initialize it marked changed.


  2. 0

    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()
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search