full code:
import sqlalchemy as sa
from sqlalchemy import orm
class Model:
session = None
class Person(object):
@staticmethod
def query():
return Model.session.query(Model.Person)
class PersonRelative(object):
@staticmethod
def query():
return Model.session.query(Model.PersonRelative)
class Resume(object):
@staticmethod
def query():
return Model.session.query(Model.Resume)
class Workplace(object):
@staticmethod
def query():
return Model.session.query(Model.Workplace)
class ResumeWorkplace(object):
@staticmethod
def query():
return Model.session.query(Model.ResumeWorkplace)
def __init__(self, engine):
"""Call me before using any of the tables or classes in the
model."""
metadata = sa.MetaData()
sm = orm.sessionmaker(autoflush=True, transactional=True,
bind=engine)
Model.session = orm.scoped_session(sm)
person_table = sa.Table('person', metadata, autoload = True,
autoload_with=engine)
person_relative_table = sa.Table('person_relative', metadata,
autoload = True, autoload_with=engine)
resume_table = sa.Table('resume', metadata, autoload = True,
autoload_with=engine)
workplace_table = sa.Table('workplace', metadata, autoload =
True, autoload_with=engine)
resume_workplace_table = sa.Table('resume_workplace',
metadata, autoload = True, autoload_with=engine)
orm.mapper(self.Person, person_table, properties = {
'relatives' : orm.relation(self.Person,
secondary=person_relative_table,
primaryjoin=person_table.c.id==person_relative_table.c.person_id,
secondaryjoin=person_relative_table.c.relative_id==person_table.c.id,
backref='followers'),
'resumes' : orm.relation(self.Resume, backref='person')
}
)
orm.mapper(self.Resume, resume_table, properties = {
'workplaces' : orm.relation(self.Workplace,
secondary=resume_workplace_table, backref='resumes')
}
)
orm.mapper(self.Workplace, workplace_table)
def commit(self):
Model.session.commit()
def save(self, obj):
Model.session.save(obj)
def save_or_update(self, obj):
Model.session.save_or_update(obj)
def flush(self):
Model.session.flush()
def delete(self, obj):
Model.session.delete(obj)
def clear(self):
Model.session.clear()
On Jun 3, 11:47 pm, "Michael Bayer" <[email protected]> wrote:
> Moshe C. wrote:
>
> > Weird, the first assertion already fails, but I am not using
> > ScopedSession.mapper. See the code in the first post.
>
> > On Jun 3, 11:28 pm, "Michael Bayer" <[email protected]> wrote:
> >> easy way to ensure things are working as expected:
>
> >> Moshe C. wrote:
>
> >> > In code:
>
> >> > t = Model.Resume()
> >> > t.id = something
>
> >> assert t not in Model.session
>
> its not a full example. no imports are illustrated including what
> orm.mapper() might be doing. There is obviously code which is setting up
> Session.mapper() or otherwise code within Resume().__init__() doing
> something similar.
>
>
>
> >> > w = Model.session.query(Model.Workplace).filter_by(id=idd).first()
>
> >> assert t not in Model.session
>
> >> > # model.save(t)
> >> > model.save_or_update(t)
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected]
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
-~----------~----~----~----~------~----~------~--~---
