[sqlalchemy] Questions about polymorphic mappers
Hi,
I have a couple of questions about polymorphic mapper inheritance. I'll
use the employee/manager/engineer example from the docs, if that's OK.
I'm also using SessionContext and assign_mapper, in case that makes a
difference.
I'd like to be able to call the base class (Employee), but have it
actually generate an instance of the correct subclass. To do this, I'm
using a metaclass which looks something like:
class EmployeeMeta(type):
def __call__(cls, kind, _fix_class=True):
if not _fix_class:
return type.__call__(cls, kind=kind)
# Use the Employee mapper's polymorphic_map to figure out
# which class to use
mappers = Employee.mapper.polymorphic_map
try:
mapper = mappers[kind]
except KeyError:
raise ValueError('No such employee type: %s '
'(valid types are %s)' %
(kind, ', '.join(mappers.keys(
cls = mapper.class_
return cls(kind=kind, _fix_class=False)
This seems to work quite well - if I call Employee(type='manager'), I
get an instance of the Manager class back. However, I don't know
whether my use of the polymorphic_map dictionary is safe - is it
considered a public attribute or a private one? Also, the first time I
use this I have to explicitly compile the mapper, which makes it feel
slightly wrong.
My second question is also related to the polymorphic_map attribute. I
have another situation where it would be useful to produce subclasses
for certain employee types, but fall back to the base class for other
types. For example, imagine there were other employee types in the
table ('secretary', 'tester' etc.), and most of them don't need an
Employee subclass. Currently, I would get a KeyError if I tried to load
any of those rows (because the type doesn't exist in the
polymorphic_map). It would be nice to be able to specify a fallback in
some way. I can see two ways of doing it:
1. Use a MapperExtension. This is probably the correct way to do it,
but it seems a bit 'heavyweight', and I can't see a (public) way to ask
the mapper to create and populate an instance of a different class, so
I would end up having to duplicate a lot of the code from the mapper (a
lot of which I don't really understand ;-)
2. Pass in a dict-like object (such as python 2.5's defaultdict) as the
_polymorphic_map argument to the mapper. This would be an ugly hack as
the _polymorphic_map argument is very definitely private. However, it
is very simple, and it would also work with the metaclass described
above.
This is probably the sort of thing which is very specific to my
situation, and SQLAlchemy shouldn't guess what I want to do, which is
why I think the MapperExtension is probably the right way to go. I just
wish it were easier to ask the mapper to use a different class.
Thanks,
Simon
--~--~-~--~~~---~--~~
You received this message because you are subscribed to the Google Groups
sqlalchemy group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~--~~~~--~~--~--~---
[sqlalchemy] Regression between r2168 and HEAD
After updating to HEAD this morning I got an error. This code has
worked with many versions of SA and I've never seen anything like
this error before. Here's the relevant code (modified from its
original version to fit your screen):
import sqlalchemy as sa
class Version(object):
table = sa.Table(version, meta,
sa.Column(id, mu.String, primary_key=True),
sa.Column(number, mu.String),
)
def assign_mapper(class_, *args, **kw):
class_.mapper = sa.mapper(class_, class_.table, *args, **kw)
assign_mapper(Version)
def verify(version, session=None):
Verify the connection and the version of the database
if session is None:
session = sa.create_session()
ver = session.query(Version).get('schema')
if ver.number != version:
raise WrongDatabaseVersionError(got %r expected %r %
(ver.number, version))
And here's the error:
Traceback (most recent call last):
File /Users/daniel/Code/PyOE/src/orderentry/app.py, line 34, in
__init__
self.dbConnect()
File /Users/daniel/Code/PyOE/src/orderentry/app.py, line 81, in
dbConnect
model.verify(__version__)
File /Users/daniel/Code/PyOE/src/orderentry/model.py, line 33,
in verify
ver = session.query(Version).get('schema')
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/query.py,
line 59, in get
return self._get(key, ident, **kwargs)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/query.py,
line 372, in _get
return self._select_statement(statement, params=params,
populate_existing=reload, version_check=(lockmode is not None))[0]
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/query.py,
line 380, in _select_statement
return self.execute(statement, params=params, **kwargs)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/query.py,
line 312, in execute
return self.instances(result, **kwargs)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/query.py,
line 331, in instances
self.mapper._instance(context, row, result)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/mapper.py,
line 1159, in _instance
identitykey = self.identity_key_from_row(row)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/orm/mapper.py,
line 750, in identity_key_from_row
return (self.class_, tuple([row[column] for column in
self.pks_by_table[self.mapped_table]]), self.entity_name)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/engine/
base.py, line 741, in __getitem__
return self.__parent._get_col(self.__row, key)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/engine/
base.py, line 634, in _get_col
rec = self._convert_key(key)
File /Users/daniel/Code/SQLAlchemy/lib/sqlalchemy/engine/
base.py, line 618, in _convert_key
raise exceptions.NoSuchColumnError(Could not locate column in
row for column '%s' % str(key))
NoSuchColumnError: Could not locate column in row for column
'version.id'
Any idea what's going on here? Thanks.
~ Daniel
--~--~-~--~~~---~--~~
You received this message because you are subscribed to the Google Groups
sqlalchemy group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~--~~~~--~~--~--~---
[sqlalchemy] Re: Test data apparently not saving
OK, i made it into a straight down program (attached) and it
completes fine.
--~--~-~--~~~---~--~~
You received this message because you are subscribed to the Google Groups
sqlalchemy group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~--~~~~--~~--~--~---
from datetime import datetime
import sqlalchemy as sa
from sqlalchemy.ext.sessioncontext import SessionContext
metadata = sa.BoundMetaData('sqlite:///:memory:')
metadata.engine.echo = True
ctx = SessionContext(sa.create_session)
groups_table = sa.Table('tg_group', metadata,
sa.Column('group_id', sa.Integer, primary_key=True),
sa.Column('group_name', sa.Unicode(16), unique=True),
sa.Column('display_name', sa.Unicode(255)),
sa.Column('created', sa.DateTime, default=datetime.now)
)
users_table = sa.Table('tg_user', metadata,
sa.Column('user_id', sa.Integer, primary_key=True),
sa.Column('user_name', sa.Unicode(16), unique=True),
sa.Column('email_address', sa.Unicode(255), unique=True),
sa.Column('display_name', sa.Unicode(255)),
sa.Column('password', sa.Unicode(40)),
sa.Column('created', sa.DateTime, default=datetime.now)
)
user_group_table = sa.Table('user_group', metadata,
sa.Column('user_id', sa.Integer, sa.ForeignKey('tg_user.user_id')),
sa.Column('group_id', sa.Integer, sa.ForeignKey('tg_group.group_id'))
)
class User(object):
def __repr__(self):
return class User user_id=%s email_address=%s display_name=%s % (
self.user_id, self.email_address, self.display_name)
class Group(object):
pass
gmapper = sa.mapper(Group, groups_table)
umapper = sa.mapper(User, users_table, properties=dict(
groups = sa.relation(Group, secondary=user_group_table, backref='users')
)
)
def sa_create(class_, **kw):
try:
obj = class_(**kw)
except TypeError:
obj = class_()
sa_update(obj, **kw) # update takes care of save/flush
return obj
def sa_update(obj, **kw):
[setattr(obj, key, value) for key, value in kw.iteritems()]
push_to_db(obj)
def push_to_db(obj):
had_session = True
session = sa.orm.session.object_session(obj)
if not session:
print Didn't find session
session = ctx.current
had_session = False
session.save(obj)
session.flush((obj,))
if not had_session:
session.expunge(obj)
ctx.current.clear()
session = ctx.current
# Create tables
metadata.drop_all()
metadata.create_all()
user1 = User()
user1.user_name='bobvilla'
user1.email_address='[EMAIL PROTECTED]'
user1.display_name='Bob Villa'
user1.password='toughasnails'
session.save(user1)
u2 = User()
u2.user_name='bobathome'
u2.email_address='[EMAIL PROTECTED]'
u2.display_name='Bob Villa'
u2.password='hammertime'
session.save(u2)
session.flush()
print 'UuUuUuU %s' % user1
user1 = user1
session.clear()
session = ctx.current
u1 = session.query(User).select(User.c.user_name=='bobvilla')
l = session.query(User).select()
print u1
for u in l:
print u
assert u1[0].user_id==user1.user_id
assert u1[0].email_address=='[EMAIL PROTECTED]'
assert len(l) == 2
u = sa_create(User, user_name='bobevans',
email_address='[EMAIL PROTECTED]',
display_name='Bob Evans',
password='secretsauce')
assert u.user_name=='bobevans'
assert u.user_id 0
On Jan 11, 2007, at 8:36 PM, Patrick Lewis wrote:
Michael Bayer wrote:
you need to attach a test case that doenst have all the generalorm
dependencies. also, i looked at this and i dont see why the
ruledispatch stuff cant be factored out of the test as well since
none of those functions seem to be called.
Sorry about that. I thought I had removed all of those, but obviously
not. Here is a one file test. I've also stripped out the dispatch
rules, but left the essential logic.
http://paste.turbogears.org/paste/831
As i mentioned before, the test_setup case passes when run by itself,
but not in combination with the test_create test. test_create always
seems to pass.
--~--~-~--~~~---~--~~
You received this message because you are subscribed to the Google
Groups sqlalchemy group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to sqlalchemy-
[EMAIL PROTECTED]
For more options, visit this group at http://groups.google.com/
group/sqlalchemy?hl=en
-~--~~~~--~~--~--~---
[sqlalchemy] Re: Saving bad data fails silently? (MySQL)
thats weird, inserting None into a PK col, even for mysql, should
fail. try a small standalone test script that just INSERTs blank
PK's into tables and see if it raises errors.
On Jan 11, 2007, at 8:36 PM, Chris Shenton wrote:
I've noticed this a few times now. Due to faulty logic in my code, or
a misunderstanding of the constraints in my schema, I create a record
through the ORM, then save() and flush() it but it doesn't get put in
the DB. This is frequently due to me setting an attribute None which
has a Nullable=False constraint.
But what's buggin' me is that it fails silently. I don't know there
was a problem inserting the record. For example, I have this table:
fault_table =
Table('fault', metadata,
Column('fault_id', Integer,
primary_key=True),
Column('system_id', Integer,ForeignKey
('system.system_id'), nullable=False),
Column('faulttrack_id', Integer,ForeignKey
('faulttrack.faulttrack_id'), nullable=True), # initially empty
Column('msgpath', String, nullable=False),
Column('subject', String, nullable=False),
Column('ts_created',DateTime,
default=func.current_timestamp()),
Column('ts_updated',DateTime,
onupdate=func.current_timestamp()),
)
If I inadvertently set a system_id to None (as below) the insert
doesn't work but no exception is thrown nor error logged:
fault2 = Fault(system_id = None, # this violates the no-
NULL restriction
msgpath = '/dev/null/UNSAVEDFAULT',
subject = 'UNSAVED FAULT',
)
self.session.save(fault2)
self.session.flush()
I turned on Echo so I could see the SQL, and everything looks fine,
but nothing is stored and no error is reported:
2007-01-11 20:30:38,922 INFO sqlalchemy.engine.base.Engine.0x..4c
BEGIN
INFO:base: BEGIN
2007-01-11 20:30:38,930 INFO sqlalchemy.engine.base.Engine.0x..4c
SELECT current_timestamp
INFO:base: SELECT current_timestamp
2007-01-11 20:30:38,930 INFO sqlalchemy.engine.base.Engine.0x..4c []
INFO:base: []
2007-01-11 20:30:38,934 INFO sqlalchemy.engine.base.Engine.0x..4c
INSERT INTO fault (system_id, faulttrack_id, msgpath, subject,
ts_created, ts_updated) VALUES (%s, %s, %s, %s, %s, %s)
INFO:base: INSERT INTO fault (system_id, faulttrack_id, msgpath,
subject, ts_created, ts_updated) VALUES (%s, %s, %s, %s, %s, %s)
2007-01-11 20:30:38,935 INFO sqlalchemy.engine.base.Engine.0x..4c
[None, None, '/dev/null/UNSAVEDFAULT', 'UNSAVED FAULT',
datetime.datetime(2007, 1, 11, 20, 30, 37), None]
INFO:base: [None, None, '/dev/null/UNSAVEDFAULT', 'UNSAVED
FAULT', datetime.datetime(2007, 1, 11, 20, 30, 37), None]
2007-01-11 20:30:38,940 INFO sqlalchemy.engine.base.Engine.0x..4c
COMMIT
INFO:base: COMMIT
So how do I detect that I'm trying to insert data that will not be
saved? Or is there some other work around I'm not aware of?
I'm using MySQL if it makes any difference, and have noticed this both
under Pylons and from a stand-alone python app like the one above.
Thanks.
--~--~-~--~~~---~--~~
You received this message because you are subscribed to the Google Groups
sqlalchemy group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~--~~~~--~~--~--~---
[sqlalchemy] Re: autoloading oracle tables and the owner parameter
i notice your line numbers arent matching up with the most recent version of SA..they seem to match up with 0.3.0 or 0.3.1 which is before this bug was fixed (in 0.3.2 - http://www.sqlalchemy.org/trac/ticket/373).see if it works with the most recent version. --~--~-~--~~~---~--~~ You received this message because you are subscribed to the Google Groups sqlalchemy group. To post to this group, send email to sqlalchemy@googlegroups.com To unsubscribe from this group, send email to [EMAIL PROTECTED] For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en -~--~~~~--~~--~--~---
[sqlalchemy] Re: Test data apparently not saving
Ok, that works for me, too. But, if I rework it how I think the test suite is running things, I get the same error. http://paste.turbogears.org/paste/832 --~--~-~--~~~---~--~~ You received this message because you are subscribed to the Google Groups sqlalchemy group. To post to this group, send email to sqlalchemy@googlegroups.com To unsubscribe from this group, send email to [EMAIL PROTECTED] For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en -~--~~~~--~~--~--~---
[sqlalchemy] Re: Test data apparently not saving
A minor revision (made user1 global) http://paste.turbogears.org/paste/833 --~--~-~--~~~---~--~~ You received this message because you are subscribed to the Google Groups sqlalchemy group. To post to this group, send email to sqlalchemy@googlegroups.com To unsubscribe from this group, send email to [EMAIL PROTECTED] For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en -~--~~~~--~~--~--~---
[sqlalchemy] How to query like sys_id=42 AND ts_created 90 minutes ago ?
I've got a bunch of history and other timestamped information I will
need to query against. The columns are created with type DateTime and
get set upon row creation:
history_table = Table(
'history', metadata,
Column('history_id', Integer,primary_key=True),
Column('system_id', Integer,ForeignKey('system.system_id'),
nullable=False),
Column('ts_created', DateTime,
default=func.current_timestamp()),
Column('ts_updated', DateTime,
onupdate=func.current_timestamp()),
)
I'm going to want to do lots of queries on this 'history' table for a
specific 'system_id' and a 'ts_created' within some duration in the
past -- like 5 or 60 minutes.
It's taken me a while to figure out the SQLAlchemy syntax to make this
work and it seems a bit verbose:
session.query(History).select(and_(History.c.system_id==42,
History.c.ts_created
datetime.datetime.now() - datetime.timedelta(minutes=90)))
Is there a better, more concise way to say this?
Thanks.
--~--~-~--~~~---~--~~
You received this message because you are subscribed to the Google Groups
sqlalchemy group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~--~~~~--~~--~--~---
