Use sqlalchemy ORM to access and update a nested field in a json column

Question

I have a table with one column containing json data. The schema of the table is like this

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    data = Column(JSON, default={})

The data in the data column has a form like this:

{
  'field_a': 1234, 
  'field_b': 5678
}

To access the data in the fields, i use the specific json functions from sqlite/SQL server and use hybrid props to have easier access. So the table looks like this.

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    data = Column(JSON, default={})

    
    @hybrid_property
    def field_a(self):
        return self.data.get('field_a')

    @field_a.setter
    def field_a(self, value):
        self.data['field_a'] = value

    @field_a.expression
    def field_a(cls):
        return func.json_extract(cls.data, '$.field_a')
    
    @field_a.update_expression
    def field_a(cls, value):
        return [
            (cls.data, func.json_set(cls.data, '$.field_a', value))
        ]

Now, i can make queries to access the data and to update the data using core and orm functions like the following:

# core query
session.execute(sa.select(User.id, User.field_a))
# core update
session.execute(sa.update(User).where(User.id == 1).values({'field_a':8888}))
# orm query
session.query(User.field_a).all()
# orm update
session.query(User).filter(User.id == 1).update({'field_a': 6666}) 

However what i would like to do.

user = session.query(User).filter(User.id==1).one()

# the update statement shall only modify field_a in data and not update the
# whole json data in the column
user.field_a = 5555
session.commit()

With the design like above this will issue an update statement of the complete data in the data column, while i would like it to issue only the partial update via func.json... functions. This poses a problem for me since i could have another process which issued an UPDATE of field_b in the time between querying and updating.

Is a structure like the one i want even possible? Actually, i am not even interested in the complete data column but just in a couple of nested field within that column.

Answer 1

Since i did not get any responses here, i went to sqlalchemy github discussions and posted the same question there (link). I got the following answer from @zzzeek:

the ORM won't do that directly and there's no hook in the unit of work that can make that happen for an ordinary change event. you need to emit the update() statement directly instead.