I have tried to create some custom SA type. And got in situation when I
can't find correct wayout. Minimal test case in attach.
My app have class SiteVersion, it can be used as regular object and as SA
mmaped object. When I use only SiteVesionDeco(see attach) all works fine,
except propagating updates to DB. When I have tried to add
SiteVersionWatch, everything failed down:
$ ~/tmp/z/z.py
2012-12-04 13:15:10,134 INFO sqlalchemy.engine.base.Engine PRAGMA
table_info("my_data")
2012-12-04 13:15:10,134 INFO sqlalchemy.engine.base.Engine ()
2012-12-04 13:15:10,134 INFO sqlalchemy.engine.base.Engine
CREATE TABLE my_data (
id INTEGER NOT NULL,
data VARCHAR,
PRIMARY KEY (id)
)
2012-12-04 13:15:10,135 INFO sqlalchemy.engine.base.Engine ()
2012-12-04 13:15:10,135 INFO sqlalchemy.engine.base.Engine COMMIT
2012-12-04 13:15:10,135 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-12-04 13:15:10,136 INFO sqlalchemy.engine.base.Engine INSERT INTO
my_data (data) VALUES (?)
2012-12-04 13:15:10,136 INFO sqlalchemy.engine.base.Engine ('1.0',)
2012-12-04 13:15:10,136 INFO sqlalchemy.engine.base.Engine COMMIT
2012-12-04 13:15:10,137 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-12-04 13:15:10,137 INFO sqlalchemy.engine.base.Engine SELECT
my_data.id AS my_data_id, my_data.data AS my_data_data
FROM my_data
WHERE my_data.id = ?
2012-12-04 13:15:10,137 INFO sqlalchemy.engine.base.Engine (1,)
2012-12-04 13:15:10,137 INFO sqlalchemy.engine.base.Engine ROLLBACK
Traceback (most recent call last):
File "/home/surabujin/tmp/z/z.py", line 172, in <module>
main()
File "/home/surabujin/tmp/z/z.py", line 169, in main
dbs.commit()
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py",
line 703, in commit
self.transaction.commit()
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py",
line 361, in commit
self._prepare_impl()
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py",
line 340, in _prepare_impl
self.session.flush()
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py",
line 1718, in flush
self._flush(objects)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py",
line 1789, in _flush
flush_context.execute()
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/unitofwork.py",
line 331, in execute
rec.execute(self)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/unitofwork.py",
line 475, in execute
uow
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/persistence.py",
line 54, in save_obj
table, states_to_update)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/persistence.py",
line 326, in _collect_update_commands
attributes.PASSIVE_NO_INITIALIZE)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/attributes.py",
line 1249, in get_state_history
return state.get_history(key, passive)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/state.py",
line 104, in get_history
return self.manager[key].impl.get_history(self, self.dict, passive)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/attributes.py",
line 533, in get_history
self, state, dict_.get(self.key, NO_VALUE))
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/orm/attributes.py",
line 1150, in from_scalar_attribute
elif attribute.is_equal(current, original) is True:
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/types.py",
line 750, in compare_values
return self.impl.compare_values(x, y)
File
"/home/surabujin/devel/DORS/env/local/lib/python2.7/site-packages/sqlalchemy/types.py",
line 84, in compare_values
return x == y
File "/home/surabujin/tmp/z/z.py", line 77, in __eq__
raise NotImplementedError('Expect %s instance, got %s' % (SiteVersion,
other))
NotImplementedError: Expect <class '__main__.SiteVersion'> instance, got
<symbol 'NO_VALUE>
Why it try to make SA specific comparison before converting object into
built-in type String? How can I fix it?
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To view this discussion on the web visit
https://groups.google.com/d/msg/sqlalchemy/-/Cdl5FTQOAE0J.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
#!/usr/bin/env python
# -*- coding:utf8 -*-
import functools
from sqlalchemy import *
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.mutable import Mutable
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import TypeDecorator, VARCHAR
@functools.total_ordering
class SiteVersion(object):
"""Parse and store version"""
version = revision = None
def __init__(self, raw):
super(SiteVersion, self).__init__()
if raw is None:
return
dst = [0 for x in xrange(2)]
v = raw.strip()
v = v.split('.')
if len(dst) < len(v):
raise ValueError(
'Invalid version string "{}". Excepcted version format N.N, where N is 0..INT_MAX'
.format(raw))
for idx in xrange(len(dst)):
try:
ent = v[idx]
try:
ent = int(ent)
except ValueError:
raise ValueError(
'Invalid version string "{}". "{}" is not numer'
.format(raw, ent))
if ent < 0:
raise ValueError(
'Version component less than 0: raw="{}", fail_idx={}, fail_component'
.format(raw, idx, ent))
dst[idx] = ent
except IndexError:
break
self.version, self.revision = dst
def __str__(self):
if self.is_undefined():
return u'n/a'
return '.'.join(str(x) for x in (self.version, self.revision))
def __repr__(self):
return '<{}: {}>'.format(self.__class__.__name__, str(self))
def __lt__(self, other):
if not isinstance(other, SiteVersion):
raise NotImplementedError('Expect %s instance, got %s' % (SiteVersion, other))
local = (self.version, self.revision)
remote = (other.version, other.revision)
for ent in zip(local, remote):
if ent[0] == ent[1]:
continue
res = ent[0] < ent[1]
break
else:
res = False
return res
def __eq__(self, other):
if not isinstance(other, SiteVersion):
raise NotImplementedError('Expect %s instance, got %s' % (SiteVersion, other))
local = (self.version, self.revision)
remote = (other.version, other.revision)
res = False
for ent in zip(local, remote):
if ent[0] != ent[1]:
break
else:
res = True
return res
reduce_json = __str__
def is_undefined(self):
return None in (self.version, self.revision)
def dup(self):
return self.__class__(str(self))
class SiteVersionDeco(TypeDecorator):
impl = String
def process_bind_param(self, value, dialect):
if value is None:
return value
if isinstance(value, basestring):
return value
if not isinstance(value, SiteVersion):
raise NotImplementedError
if value.is_undefined():
return None
return str(value)
def process_result_value(self, value, dialect):
return SiteVersion(value)
class _SiteVersionWatch(Mutable, SiteVersion):
@classmethod
def coerce(cls, attr, val):
if isinstance(val, _SiteVersionWatch):
pass
elif val is None:
val = _SiteVersionWatch(None)
elif isinstance(val, SiteVersion):
o = cls(None)
o.version = val.version
o.revision = val.revision
val = o
elif isinstance(val, basestring):
val = cls(val)
else:
val = Mutale.coerce(attr, val) # raise ValueError
return val
def __setattr__(self, attr, val):
super(_SiteVersionWatch, self).__setattr__(attr, val)
if attr in ('vversion', 'revision'):
self.changed()
_SiteVersionWatch.associate_with(SiteVersionDeco)
Base = declarative_base()
class MyDataClass(Base):
__tablename__ = 'my_data'
id = Column(Integer, primary_key=True)
data = Column(SiteVersionDeco)
def main():
engine = create_engine('sqlite://', echo=True)
Base.metadata.create_all(engine)
# create a configured "Session" class
Session = sessionmaker(bind=engine)
# create a Session
dbs = Session()
# work with sess
obj = MyDataClass(data='1.0')
dbs.add(obj)
dbs.commit()
obj.data.revision = 1
dbs.commit()
if __name__ == '__main__':
main()
