[Sqlalchemy-tickets] Issue #4303: Union over enum field raises LookUpError (zzzeek/sqlalchemy)

[Patrick v. d. L. <iss...@bi...>]

New issue 4303: Union over enum field raises LookUpError
https://bitbucket.org/zzzeek/sqlalchemy/issues/4303/union-over-enum-field-raises-lookuperror

Patrick van der Leer:

There seems to be a bug in the union select over a enum field

    import enum
    from datetime import datetime

    import sqlalchemy as db

    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.ext.hybrid import hybrid_property
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()


    class MyModel(Base):
        __abstract__ = True
        def save(self, commit=True):
            """Save the record."""
            session.add(self)
            if commit:
                session.commit()
            return self
        

    class Setting(MyModel):
        __tablename__ = "setting"
        
        id = db.Column(db.Integer, primary_key=True, autoincrement=True)
        key = db.Column(db.String(240), nullable=False)

        type_value = db.Column(db.Unicode(16))
        int_value = db.Column(db.Integer, info={'type': (int, 'integer')})
        float_value = db.Column(db.Float, info={'type': (float, 'float')})
        char_value = db.Column(db.UnicodeText, info={'type': (str, 'string')})
        boolean_value = db.Column(db.Boolean, info={'type': (bool, 'boolean')})
        datetime_value = db.Column(db.DateTime, info={'type': (datetime, 'datetime')})

        @hybrid_property
        def value(self):
            if self.type_value and self.type_value in self.type_map.keys():
                column, discriminator = self.type_map[self.type_value]
                if column is None:
                    return None
                return getattr(self, column)
            return None

        @value.setter
        def value(self, v):
            _type = type(v)
            if v is not None and _type in self.type_map.keys():
                column, discriminator = self.type_map[_type]
                setattr(self, column, v)
                self.type_value = discriminator
            else:
                raise TypeError("Unable to save value of type: %s" % str(type(v)))

        type = db.Column(db.String(50))
        __mapper_args__ = {
            'polymorphic_identity': 'setting',
            'polymorphic_on': type
        }


    @db.event.listens_for(Setting, "mapper_configured", propagate=True)
    def on_new_class(mapper, cls_):
        """Look for Column objects with type info in them, and work up
        a lookup table."""

        info_dict = {
            type(None): (None, 'none'),
            'none': (None, 'none')
        }

        for k in mapper.c.keys():
            col = mapper.c[k]
            if 'type' in col.info:
                python_type, discriminator = col.info['type']
                info_dict[python_type] = (k, discriminator)
                info_dict[discriminator] = (k, discriminator)
        cls_.type_map = info_dict


    class SettingAppliesTo(enum.Enum):
        default = "default"
        company = "company"
        project = "project"
        user = "user"


    class DefaultSetting(Setting):
        applies_to = db.Column(db.Enum(SettingAppliesTo))

        __mapper_args__ = {
            'polymorphic_identity': 'global_setting',
        }


    class CompanySetting(Setting):
        #company_id = db.Column(db.ForeignKey('company.id'))
        # company = db.relationship("Company")
        
        company_id = db.Column(db.Integer)
        
        __mapper_args__ = {
            'polymorphic_identity': 'company_setting',
        }

        @classmethod
        def get(cls, company_id: int, applies_to=SettingAppliesTo.company):
            ds = session.query(DefaultSetting).filter_by(applies_to=applies_to)
            cs = session.query(CompanySetting).filter_by(company_id=company_id)
            settings = ds.union(cs)
            return settings
            
    engine = create_engine('sqlite:///:memory:', echo=True)
    Session = sessionmaker(bind=engine)
    session = Session()

    Base.metadata.create_all(engine)
    MyModel.metadata.create_all(engine)

            
    DefaultSetting(
        key="company.public",
        value=False,
        applies_to=SettingAppliesTo.company
    ).save()

    DefaultSetting(
        key="company.test",
        value=False,
        applies_to=SettingAppliesTo.company
    ).save()


    CompanySetting(
        key="company.public",
        value=True,
        company_id=1
    ).save()

    company_settings = list(CompanySetting.get(1))

    
    
This raises the following error:
    
    Traceback (most recent call last):
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/sql/sqltypes.py", line 1424, in _object_value_for_elem
        return self._object_lookup[elem]
    KeyError: 1

    During handling of the above exception, another exception occurred:

    Traceback (most recent call last):
      File "/usr/local/lib/python3.6/dist-packages/unittest2/case.py", line 67, in testPartExecutor
        yield
      File "/usr/local/lib/python3.6/dist-packages/unittest2/case.py", line 625, in run
        testMethod()
      File "/home/patrick/PycharmProjects/regripp-backend/tests/models/test_setting.py", line 82, in test_company_settings
        company_settings = list(CompanySetting.get(company.id))
      File "/home/patrick/PycharmProjects/regripp-backend/Regripp/models/settings.py", line 97, in get
        ff = list(settings)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/orm/loading.py", line 98, in instances
        util.raise_from_cause(err)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/util/compat.py", line 265, in raise_from_cause
        reraise(type(exception), exception, tb=exc_tb, cause=cause)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/util/compat.py", line 249, in reraise
        raise value
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/orm/loading.py", line 79, in instances
        rows = [proc(row) for row in fetch]
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/orm/loading.py", line 79, in <listcomp>
        rows = [proc(row) for row in fetch]
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/orm/loading.py", line 743, in polymorphic_instance
        return _instance(row)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/orm/loading.py", line 511, in _instance
        loaded_instance, populate_existing, populators)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/orm/loading.py", line 611, in _populate_full
        dict_[key] = getter(row)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/sql/sqltypes.py", line 1507, in process
        value = self._object_value_for_elem(value)
      File "/usr/local/lib/python3.6/dist-packages/sqlalchemy/sql/sqltypes.py", line 1427, in _object_value_for_elem
        '"%s" is not among the defined enum values' % elem)
    LookupError: "1" is not among the defined enum values
    
    
Now looking at the function:

    def _object_value_for_elem(self, elem):
        try:
            return self._object_lookup[elem]
        except KeyError:
            raise LookupError(
                '"%s" is not among the defined enum values' % elem)

While setting a breakpoint on the return I see that the value of elem changes from a str `company` to int 1


Tested on sqlalchemy version 1.2.5 and 1.2.9