Python – SQLAlchemy enumeration in external files

I have a Postgres database with a table of pending operations. One of the operations in an enumeration has the status of the enumeration. I used the standard python (2.7) enumeration and AutoNumber (myenum.py):

class AutoNumber(enum. Enum):
    def __new__(cls):
        value = len(cls.__members__) + 1
        obj = object.__new__(cls)
        obj._value_ = value
        return obj

class MyStatus(AutoNumber):

INITIAL = ()
    ACCEPTED = ()
    DENIED = ()
    ACK_PENDING = ()
    AUTHORIZED = ()
    ACTIVE = ()
    END = ()
    DELETED = ()
# end enum

The table looks like (also in myenum.py

):

Base = declarative_base()

class MyOperation(Base):

__tablename__ = 'operations'

id  = Column( Integer, primary_key=True )

status = Column( Enum(MyStatus) )
    status_message = Column( String )
    status_time = Column( DateTime )

def __repr__(self):
        return "<MyOperation(%s, %s, %s, %s)>" % \
            ( self.id, self.status, self.status_time, self.status_message )
# end class

Usually this works well. In the same file where MyStatus (myoper.py) is defined, I can change the state and save it back to the database and it works fine :

def checkOper( oper ):
    oper.status = MyStatus.DENIED
    oper.status_message = "failed check (internal)"
    oper.status_time = datetime.datetime.utcnow()

That’s how I called it (in myoper.py).

    checkOper( oper )
    session.add(oper)
    session.commit()

It’s all in the same file (myoper.py).

However, if I pass an oper object to an external function, and IT changes state, then I get a sqlalchemy.exc.StatementError.

This is the external function (myoper_test.py):

import datetime
from myoper import MyStatus

def extCheckOper( oper ):
    oper.status = MyStatus.DENIED
    oper.status_message = "failed check (external)"
    oper.status_time = datetime.datetime.utcnow()

That’s how I called it (from myoper.py):

    from myoper_test import extCheckOper
    extCheckOper( oper )
    session.add(oper)
    session.commit()

Here is the stack trace:

Traceback (most recent call last):
  File "./myoper.py", line 120, in <module>
    session.commit()
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 906, in commit
    self.transaction.commit()
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 461, in commit
    self._prepare_impl()
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 441, in _prepare_impl
    self.session.flush()
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 2177, in flush
    self._flush(objects)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 2297, in _flush
    transaction.rollback(_capture_exception=True)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/util/langhelpers.py", line 66, in __exit__
    compat.reraise(exc_type, exc_value, exc_tb)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 2261, in _flush
    flush_context.execute()
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/unitofwork.py", line 389, in execute
    rec.execute(self)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/unitofwork.py", line 548, in execute
    uow
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/persistence.py", line 177, in save_obj
    mapper, table, update)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/persistence.py", line 737, in _emit_update_statements
    execute(statement, multiparams)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 945, in execute
    return meth(self, multiparams, params)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/sql/elements.py", line 263, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 1053, in _execute_clauseelement
    compiled_sql, distilled_params
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 1121, in _execute_context
    None, None)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 1402, in _handle_dbapi_exception
    exc_info
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/util/compat.py", line 203, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 1116, in _execute_context
    context = constructor(dialect, self, conn, *args)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/default.py", line 639, in _init_compiled
    for key in compiled_params
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/engine/default.py", line 639, in <genexpr>
    for key in compiled_params
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/sql/sqltypes.py", line 1446, in process
    value = self._db_value_for_elem(value)
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/sql/sqltypes.py", line 1354, in _db_value_for_elem
    '"%s" is not among the defined enum values' % elem)
sqlalchemy.exc.StatementError: (exceptions. LookupError) "MyStatus.DENIED" is not among the defined enum values [SQL: u'UPDATE operations SET status=%(status)s, status_message=%(status_message)s, status_time=%(status_time)s WHERE operations.id = %(operations_id)s'] [parameters: [{'status': <MyStatus.DENIED: 6>, 'status_time': datetime.datetime(2017, 10, 18, 20, 22, 44, 350035), 'status_message': ' failed check (external)', 'operations_id': 3}]]

I’ve tried checking the types in both the internal and external files, but it’s listed in the way <enum 'MyStatus'>

I

found that if I assign oper.status to the enumeration .name, that does work:

def extCheckOper( oper ):
    oper.status = MyStatus.AUTHORIZED.name
    oper.status_message = "authorized check (external)"
    oper.status_time = datetime.datetime.utcnow()

But this is obviously ugly.

So – what am I doing wrong? How does MyStatus differ from an external file that messes up SQL Alchemy in its defined file?

I posted this question to the SQL Alchemy mailing list and got the answer. Link to thread

It turns out that this is one of those “intractable problems” about python and has nothing to do with SQL Alchemy. Here is a quote: Executing Main Module Twice .

In this particular case, when I execute my script, MyStatus is created with a specific id (python handle of the type). But when myoper_test imports MyStatus from myoper, it creates a different id.

Therefore, when

extCheckOper assigns a MyStatus value to a status field, it is different from the MyStatus that SQL Alchemy uses when creating the DB mapping, so when SQL Alchemy tries to save it to the DB, the “is” operator fails because (external) MyStatus is not the same as (raw) MyStatus.

There are several different workarounds. One way is not to run the code as main (after moving the exited main code to the main() function:

$ python -c "from myoper import main; import sys; main(*sys.argv[1:])" ext_check 1

A better solution is to avoid this problem altogether – move the externally called code into the internal test script. The code in main is mostly kept in the main script (sorry, can’t resist…:-)).