assert sess.query(Employee).\
get([c2.company_id, 3]).reports_to.name == 'emp5'
-class ComplexPostUpdateTest(fixtures.MappedTest):
- @classmethod
- def define_tables(cls, metadata):
- Table("jobs", metadata,
- Column("jobno", sa.Unicode(15), primary_key=True),
- Column("created", sa.DateTime, nullable=False,
- default=datetime.datetime.now),
- Column("deleted", sa.Boolean, nullable=False, default=False))
-
- Table("pageversions", metadata,
- Column("jobno", sa.Unicode(15), primary_key=True),
- Column("pagename", sa.Unicode(30), primary_key=True),
- Column("version", Integer, primary_key=True, default=1),
- Column("created", sa.DateTime, nullable=False,
- default=datetime.datetime.now),
- Column("md5sum", String(32)),
- Column("width", Integer, nullable=False, default=0),
- Column("height", Integer, nullable=False, default=0),
- sa.ForeignKeyConstraint(
- ["jobno", "pagename"],
- ["pages.jobno", "pages.pagename"]))
-
- Table("pages", metadata,
- Column("jobno", sa.Unicode(15), ForeignKey("jobs.jobno"),
- primary_key=True),
- Column("pagename", sa.Unicode(30), primary_key=True),
- Column("created", sa.DateTime, nullable=False,
- default=datetime.datetime.now),
- Column("deleted", sa.Boolean, nullable=False, default=False),
- Column("current_version", Integer))
-
- Table("pagecomments", metadata,
- Column("jobno", sa.Unicode(15), primary_key=True),
- Column("pagename", sa.Unicode(30), primary_key=True),
- Column("comment_id", Integer, primary_key=True,
- autoincrement=False),
- Column("content", sa.UnicodeText),
- sa.ForeignKeyConstraint(
- ["jobno", "pagename"],
- ["pages.jobno", "pages.pagename"]))
-
- @classmethod
- def setup_mappers(cls):
- pageversions, pagecomments, jobs, pages = (cls.tables.pageversions,
- cls.tables.pagecomments,
- cls.tables.jobs,
- cls.tables.pages)
-
- class Job(cls.Basic):
- def create_page(self, pagename):
- return Page(job=self, pagename=pagename)
- class PageVersion(cls.Basic):
- def __init__(self, page=None, version=None):
- self.page = page
- self.version = version
- class Page(cls.Basic):
- def __init__(self, job=None, pagename=None):
- self.job = job
- self.pagename = pagename
- self.currentversion = PageVersion(self, 1)
- def add_version(self):
- self.currentversion = PageVersion(
- page=self, version=self.currentversion.version+1)
- comment = self.add_comment()
- comment.closeable = False
- comment.content = u'some content'
- return self.currentversion
- def add_comment(self):
- nextnum = max([-1] +
- [c.comment_id for c in self.comments]) + 1
- newcomment = PageComment()
- newcomment.comment_id = nextnum
- self.comments.append(newcomment)
- newcomment.created_version = self.currentversion.version
- return newcomment
- class PageComment(cls.Basic):
- pass
-
- mapper(Job, jobs)
- mapper(PageVersion, pageversions)
- mapper(Page, pages, properties={
- 'job': relationship(
- Job,
- backref=backref('pages',
- cascade="all, delete-orphan",
- order_by=pages.c.pagename)),
- 'currentversion': relationship(
- PageVersion,
- uselist=False,
- primaryjoin=sa.and_(
- pages.c.jobno==pageversions.c.jobno,
- pages.c.pagename==pageversions.c.pagename,
- pages.c.current_version==pageversions.c.version),
- post_update=True),
- 'versions': relationship(
- PageVersion,
- cascade="all, delete-orphan",
- primaryjoin=sa.and_(pages.c.jobno==pageversions.c.jobno,
- pages.c.pagename==pageversions.c.pagename),
- order_by=pageversions.c.version,
- backref=backref('page',lazy='joined')
- )})
- mapper(PageComment, pagecomments, properties={
- 'page': relationship(
- Page,
- primaryjoin=sa.and_(pages.c.jobno==pagecomments.c.jobno,
- pages.c.pagename==pagecomments.c.pagename),
- backref=backref("comments",
- cascade="all, delete-orphan",
- order_by=pagecomments.c.comment_id))})
-
- def test_basic(self):
- """A combination of complicated join conditions with post_update."""
-
- Job = self.classes.Job
-
-
- j1 = Job(jobno=u'somejob')
- j1.create_page(u'page1')
- j1.create_page(u'page2')
- j1.create_page(u'page3')
-
- j2 = Job(jobno=u'somejob2')
- j2.create_page(u'page1')
- j2.create_page(u'page2')
- j2.create_page(u'page3')
-
- j2.pages[0].add_version()
- j2.pages[0].add_version()
- j2.pages[1].add_version()
-
- s = create_session()
- s.add_all((j1, j2))
-
- s.flush()
-
- s.expunge_all()
- j = s.query(Job).filter_by(jobno=u'somejob').one()
- oldp = list(j.pages)
- j.pages = []
-
- s.flush()
-
- s.expunge_all()
- j = s.query(Job).filter_by(jobno=u'somejob2').one()
- j.pages[1].current_version = 12
- s.delete(j)
- s.flush()
class FKsAsPksTest(fixtures.MappedTest):
"""Syncrules on foreign keys that are also primary"""