I have Postgres DB with a table of pending operations. One column in the
operation in an enum with the status of the enum. I used the standard
python (2.7) enum, with 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
Generally this works fine. In the SAME FILE that defines MyStatus
(myoper.py), I can change the status and save it back to the DB and it
works fine:
def checkOper( oper ):
oper.status = MyStatus.DENIED
oper.status_message = "failed check (internal)"
oper.status_time = datetime.datetime.utcnow()
Here's how I call it (within myoper.py)
checkOper( oper )
session.add(oper)
session.commit()
This is all in the same file (myoper.py).
However, if I pass an oper instance to an external function, and that
function changes the status, then I get a sqlalchemy.exc.StatementError.
Here's the external function (myoper_test.py):
import datetimefrom myoper import MyStatus
def extCheckOper( oper ):
oper.status = MyStatus.DENIED
oper.status_message = "failed check (external)"
oper.status_time = datetime.datetime.utcnow()
Here's how I call it (from myoper.py):
from myoper_test import extCheckOper
extCheckOper( oper )
session.add(oper)
session.commit()
Here's 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 inspecting the type both in the internal file, and external
file, but it's always listed as <enum 'MyStatus'>.
I have found, that if I assign the oper.status to the enum .name, then that
DOES work:
def extCheckOper( oper ):
oper.status = MyStatus.AUTHORIZED.name
oper.status_message = "authorized check (external)"
oper.status_time = datetime.datetime.utcnow()
But that's obviously not a good solution.
So - what am I doing wrong? What is different about MyStatus in the
myoper.py, vs an external file (myoper_test.py) that screws up SQL Alchemy?
I've attached the files I've been testing with:
[~/test] python ./myoper.py drop
dropping table
[~/test] python ./myoper.py
dumping DB
[~/test] python ./myoper.py new
committing oper: <MyOperation(None, MyStatus.ACCEPTED, 2017-10-18
22:14:24.027276, Initial)>
after commit: <MyOperation(1, MyStatus.ACCEPTED, 2017-10-18
22:14:24.027276, Initial)>
[~/test] python ./myoper.py
dumping DB
<MyOperation(1, MyStatus.ACCEPTED, 2017-10-18 22:14:24.027276, Initial)>
[~/test] python ./myoper.py int_check 1
test for opid 1
found oper: <MyOperation(1, MyStatus.ACCEPTED, 2017-10-18 22:14:24.027276,
Initial)>
<enum 'MyStatus'>
after compliance check: <MyOperation(1, MyStatus.DENIED, 2017-10-18
22:14:32.246124, failed check (internal))>
status: failed check (internal)
after commit: <MyOperation(1, MyStatus.DENIED, 2017-10-18 22:14:32.246124,
failed check (internal))>
[~/test] python ./myoper.py
dumping DB
<MyOperation(1, MyStatus.DENIED, 2017-10-18 22:14:32.246124, failed check
(internal))>
[~/test] python ./myoper.py ext_check 1
test for opid 1
found oper: <MyOperation(1, MyStatus.DENIED, 2017-10-18 22:14:32.246124,
failed check (internal))>
<enum 'MyStatus'>
after compliance check: <MyOperation(1, MyStatus.AUTHORIZED, 2017-10-18
22:14:40.992209, authorized check (external))>
status: authorized check (external)
Traceback (most recent call last):
File "./myoper.py", line 120, in <module>
session.commit()
... <stacktrace as above> ...
[~/test] python ./myoper.py
dumping DB
<MyOperation(1, MyStatus.DENIED, 2017-10-18 22:14:32.246124, failed check
(internal))>
Ubuntu 16.04, x64 - Python 2.7 - SQL Alchemy 1.1.14
Same thing happens under Cygwin, FWIW.
Thanks,
Taz
PS: if there is anything else I'm doing wrong, please let me know - I'm new to
SQL/SQLAlchemy.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
#!/usr/bin/python
import datetime
import enum
from sqlalchemy import Column, Integer, String, DateTime, Enum
from sqlalchemy.ext.declarative import declarative_base
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 = ()
USER_AUTH_ACK_PENDING = ()
AUTHORIZED = ()
ACTIVE = ()
END = ()
DELETED = ()
# end of MyStatus
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
def checkOper( oper ):
oper.status = MyStatus.DENIED
oper.status_message = "failed check (internal)"
oper.status_time = datetime.datetime.utcnow()
if __name__ == '__main__':
import os
sql_echo = 'SQL_ECHO' in os.environ
from sqlalchemy import create_engine
engine = create_engine('postgresql://localhost', echo=sql_echo)
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine, autoflush=False)
session = Session()
Base.metadata.create_all(engine)
import sys
if len(sys.argv) < 2:
print "dumping DB"
for oper in session.query(MyOperation):
print oper
sys.exit(0)
elif sys.argv[1].lower() == "drop":
print "dropping table"
# not sure why I need to pass engine here - it should have been bound
# above when I called Base.metadat.create_all
MyOperation.__table__.drop(engine)
elif sys.argv[1].lower() == "new":
# create it and add it to db
oper = MyOperation()
oper.status = MyStatus.ACCEPTED
oper.status_message = "Initial"
oper.status_time = datetime.datetime.utcnow()
print "committing oper:", oper
session.add(oper)
session.commit()
print "after commit:", oper
sys.exit(0)
elif sys.argv[1].lower() == "ext_check":
if len(sys.argv) < 3:
print "need <id>"
sys.exit(-1)
opid=sys.argv[2]
print "test for opid", opid
oper = session.query(MyOperation).filter( MyOperation.id==opid ).first()
if not oper:
print "no oper for opid:", opid
sys.exit(-1)
print "found oper:", oper
from myoper_test import extCheckOper
extCheckOper( oper )
print type( oper.status )
print "after compliance check:", oper
print "status:", oper.status_message
session.add(oper)
session.commit()
print "after commit:", oper
sys.exit(0)
elif sys.argv[1].lower() == "int_check":
if len(sys.argv) < 3:
print "need <opid>"
sys.exit(-1)
opid=sys.argv[2]
print "test for opid", opid
oper = session.query(MyOperation).filter( MyOperation.id==opid ).first()
if not oper:
print "no oper for opid:", opid
sys.exit(-1)
print "found oper:", oper
checkOper( oper )
print type( oper.status )
print "after compliance check:", oper
print "status:", oper.status_message
session.add(oper)
session.commit()
print "after commit:", oper
sys.exit(0)
elif sys.argv[1].lower() == "del":
print "delete for opid ", sys.argv[2]
opid = sys.argv[2]
entry = session.query(MyOperation).filter( MyOperation.id==opid ).first()
if entry:
session.delete( entry )
session.commit()
#!/usr/bin/python
import datetime
from myoper import MyStatus
def extCheckOper( oper ):
oper.status = MyStatus.AUTHORIZED
oper.status_message = "authorized check (external)"
oper.status_time = datetime.datetime.utcnow()
