use objectify to persist relation

[wolfgang-302]

I could not get colanderalchemy to persist data from relations. I had a look at the source of

'schema.objectify()'

where schema is an instance of 'SQLAlchemySchemaNode'

I think i found three bugs in this code:
line 691:

cls = prop.mapper.class_

here cls is created but never used.

line 694/695:

value = [self[attr].children[0].objectify(obj)
              for obj in dict_[attr]]

if objectify is called on an appstruct, where the corresponding value is already persisted,
this fails. objectify needs an context in this case.

The same holds true for
line 698:

value = self[attr].objectify(dict_[attr])

I tried to fix this. For this i redefined SQLAlchemySchemaNode:

import colander
import colanderalchemy
import deform
import logging

from sqlalchemy import inspect

log = logging.getLogger(__name__)

class Custom_colanderalchemy_SchemaNode(colanderalchemy.SQLAlchemySchemaNode):
    '''only redefine objectify on this class'''
    
    def objectify(self, dict_, context=None):
        mapper = self.inspector
        
        # if context exists, there is usually a session
        # to which it is assigned. Get this session:
        if context:
            state = inspect(context)
            session = state.session if state.session else None
        else:
            session = None
            context = mapper.class_()
        
        for attr in dict_:
            if mapper.has_property(attr):
                prop = mapper.get_property(attr)
                if hasattr(prop, 'mapper'):
                    # this is a relation?
                    
                    # get the primary_key of the class
                    # that belongs to attr
                    # todo: handle class without primary_key
                    cls = prop.mapper.class_
                    pk = [v.key for v in inspect(cls).primary_key]

                    if prop.uselist:
                        # Sequence of objects
                        
                        # all children have the same subschema
                        subschema = self[attr].children[0]
                        value = []
                        for obj in dict_[attr]:
                            # get the identity from appstruct
                            obj_identity = obj.get(*pk, None)
                            if obj_identity:
                                # this obj exists in session
                                # we need to objectify with the
                                # corresponding obj_context
                                obj_context = session.query(cls).get(obj_identity)
                            else:
                                # create new instance of related object
                                # to add to session
                                obj_context = None
                            value.append(subschema.objectify(obj, obj_context))
                    else:
                        # Single object
                        subschema = self[attr]
                        # get the identity from appstruct
                        obj_identity = obj.get(*pk, None)
                        if obj_identity:
                            # this obj exists in session
                            # we need to objectify with the
                            # corresponding obj_context
                            obj_contex = session.query(cls).get(obj_identity)
                        else:
                            # create new instance of related object
                            # to add to session
                            obj_context = None
                        value = subschema.objectify(dict_[attr], obj_contex)
                else:
                     value = dict_[attr]
                     if value is colander.null:
                         # `colander.null` is never an appropriate
                         #  value to be placed on an SQLAlchemy object
                         #  so we translate it into `None`.
                         value = None
                setattr(context, attr, value)
            else:
                # Ignore attributes if they are not mapped
                log.debug(
                    'SQLAlchemySchemaNode.objectify: %s not found on '
                    '%s. This property has been ignored.',
                    attr, self
                )
                continue

        return context

I am not really experienced in writing code (and writing bugreports). Maybe someone who knows this stuff better, can take a look at it. This is not very good tested, but it works for me. I only tried the case with a one-column primary key. Maybe this needs corretion for multiple-column primary keys.

Wolfgang

[tisdall]

#92 (comment)

I'm waiting for someone to write a PR with a test case.

[wolfgang-302]

I just had a look at
Query.get(ident)
syntax from sqlalchemy. The ident is a scalar value or a tuple. That means, the lines

                    pk = [v.key for v in inspect(cls).primary_key]

should be replaced with

                    pk = tuple(v.key for v in inspect(cls).primary_key)
                    if len(pk) == 1:
                        pk = pk[0]

Then, extracting the primary key from the appstruct is done as

                            # get the identity from appstruct
                            if isinstance(pk, tuple):
                                obj_identity = tuple(obj.get(v) for v in pk)
                            else:
                                # pk is scalar
                                obj_identity = obj.get(pk, None)

The class is then:

class Custom_colanderalchemy_SchemaNode(colanderalchemy.SQLAlchemySchemaNode):
    '''only redefine objectify on this class'''
    
    def objectify(self, dict_, context=None):
        mapper = self.inspector
        
        # if context exists, there is usually a session
        # to which it is assigned. Get this session:
        if context:
            state = inspect(context)
            session = state.session if state.session else None
        else:
            session = None
            context = mapper.class_()
        
        for attr in dict_:
            if mapper.has_property(attr):
                prop = mapper.get_property(attr)
                if hasattr(prop, 'mapper'):
                    # this is a relation?
                    
                    # get the primary_key of the class
                    # that belongs to attr
                    # todo: handle class without primary_key
                    #       test class with composite primary key
                    cls = prop.mapper.class_
                    pk = tuple(v.key for v in inspect(cls).primary_key)
                    if len(pk) == 1:
                        # scalar primary key
                        pk = pk[0]

                    if prop.uselist:
                        # Sequence of objects
                        
                        # all children have the same subschema
                        subschema = self[attr].children[0]
                        value = []
                        for obj in dict_[attr]:
                            # get the identity from appstruct
                            if isinstance(pk, tuple):
                                obj_identity = tuple(obj.get(v) for v in pk)
                            else:
                                # pk is scalar
                                obj_identity = obj.get(pk, None)
                                
                            if obj_identity:
                                # this obj exists in session
                                # we need to objectify with the
                                # corresponding obj_context
                                obj_context = session.query(cls).get(obj_identity)
                            else:
                                # create new instance of related object
                                # to add to session
                                obj_context = None
                            value.append(subschema.objectify(obj, obj_context))
                    else:
                        # Single object
                        subschema = self[attr]
                        # get the identity from appstruct
                        obj_identity = obj.get(pk, None)
                        if obj_identity:
                            # this obj exists in session
                            # we need to objectify with the
                            # corresponding obj_context
                            obj_contex = session.query(cls).get(obj_identity)
                        else:
                            # create new instance of related object
                            # to add to session
                            obj_context = None
                        value = subschema.objectify(dict_[attr], obj_contex)
                else:
                     value = dict_[attr]
                     if value is colander.null:
                         # `colander.null` is never an appropriate
                         #  value to be placed on an SQLAlchemy object
                         #  so we translate it into `None`.
                         value = None
                setattr(context, attr, value)
            else:
                # Ignore attributes if they are not mapped
                log.debug(
                    'SQLAlchemySchemaNode.objectify: %s not found on '
                    '%s. This property has been ignored.',
                    attr, self
                )
                continue

        return context

I am sorry, i had to edit this comment again. I could not really test this code for composite primary keys of the relation but i hope this works now. This must really be tested by someone who knows what to look for. With a scalar primary key of the relation it works.

Wolfgang

[tisdall]

... did you read the comment I referenced in the other issue?

[wolfgang-302]

you mean issue 92?
I found it after i opended this issue, but I think it is essentially the same problem. But

                    cls = prop.mapper.class_
                    if prop.uselist:
                        # Sequence of objects
                        value = [self[attr].children[0].objectify(obj, cls)  # added context here
                                 for obj in dict_[attr]]
                    else:
                        # Single object
                        value = self[attr].objectify(dict_[attr], cls)  # added context here

is not sufficient. context needs to be a reference to an existing instance in session. But cls is a class and not an instance and is not tied to session at all. To find the instance in session that corresponds to obj -if it exists- you need data from obj to identify this instance. To do this, i extract the primary key from cls and use this to extract the identity of obj in session. (This works only if a primary key exists).

session.query(cls).get(identity)

then returns this instance or None. None means, a new instance is created and added to session.

This solution works for me for a single column primary key. I did not test it for composite primary keys, but (I hope) the only difference is, the key is a tuple in this case.

I hope, i did not missunderstand you. Maybe you can give me some hints on what to do next.

I have really no experience in testing, if this works in any situation. This should be done by someone with more experience in this field.

Wolfgang

[tisdall]

Unfortunately I haven't worked on this library in a long time and don't currently use it. So, it doesn't make too much sense for me to spend too much time on this. I also never used relations due to other issues, too.

To submit a PR, fork this repo and check it out on your computer. Then copy a test in /tests/test_schema.py and make a base case that fails with the current code ( #92 has an example ). After that, you modify the code in /colanderalchemy/schema.py to fix the bug and then run the tests again. (python setup.py test runs the tests) Once you have it fixed, you commit it back to your forked version of the repository and then submit a pull request (PR) to be merged into this project.

[wolfgang-302]

Ok. So i forked the repo, created a branch 'persist_relation' on the forked repo and corrected the problem there.
tests/test_persist_data.py failes on branch master
all tests pass on branch 'persist_relation'.

Hope it works.
Wolfgang