"""A legacy test for a particular somewhat complicated mapping."""
from sqlalchemy import CHAR
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import Text
from sqlalchemy.orm import backref
from sqlalchemy.orm import polymorphic_union
from sqlalchemy.orm import relationship
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
class MagazineTest(fixtures.MappedTest):
@classmethod
def setup_classes(cls):
Base = cls.Comparable
class Publication(Base):
pass
class Issue(Base):
pass
class Location(Base):
pass
class LocationName(Base):
pass
class PageSize(Base):
pass
class Magazine(Base):
pass
class Page(Base):
pass
class MagazinePage(Page):
pass
class ClassifiedPage(MagazinePage):
pass
@classmethod
def define_tables(cls, metadata):
Table(
"publication",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(45), default=""),
)
Table(
"issue",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("publication_id", Integer, ForeignKey("publication.id")),
Column("issue", Integer),
)
Table(
"location",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("issue_id", Integer, ForeignKey("issue.id")),
Column("ref", CHAR(3), default=""),
Column(
"location_name_id", Integer, ForeignKey("location_name.id")
),
)
Table(
"location_name",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(45), default=""),
)
Table(
"magazine",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("location_id", Integer, ForeignKey("location.id")),
Column("page_size_id", Integer, ForeignKey("page_size.id")),
)
Table(
"page",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("page_no", Integer),
Column("type", CHAR(1), default="p"),
)
Table(
"magazine_page",
metadata,
Column(
"page_id", Integer, ForeignKey("page.id"), primary_key=True
),
Column("magazine_id", Integer, ForeignKey("magazine.id")),
Column("orders", Text, default=""),
)
Table(
"classified_page",
metadata,
Column(
"magazine_page_id",
Integer,
ForeignKey("magazine_page.page_id"),
primary_key=True,
),
Column("titles", String(45), default=""),
)
Table(
"page_size",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("width", Integer),
Column("height", Integer),
Column("name", String(45), default=""),
)
def _generate_data(self):
(
Publication,
Issue,
Location,
LocationName,
PageSize,
Magazine,
Page,
MagazinePage,
ClassifiedPage,
) = self.classes(
"Publication",
"Issue",
"Location",
"LocationName",
"PageSize",
"Magazine",
"Page",
"MagazinePage",
"ClassifiedPage",
)
london = LocationName(name="London")
pub = Publication(name="Test")
issue = Issue(issue=46, publication=pub)
location = Location(ref="ABC", name=london, issue=issue)
page_size = PageSize(name="A4", width=210, height=297)
magazine = Magazine(location=location, size=page_size)
ClassifiedPage(magazine=magazine, page_no=1)
MagazinePage(magazine=magazine, page_no=2)
ClassifiedPage(magazine=magazine, page_no=3)
return pub
def _setup_mapping(self, use_unions, use_joins):
(
Publication,
Issue,
Location,
LocationName,
PageSize,
Magazine,
Page,
MagazinePage,
ClassifiedPage,
) = self.classes(
"Publication",
"Issue",
"Location",
"LocationName",
"PageSize",
"Magazine",
"Page",
"MagazinePage",
"ClassifiedPage",
)
self.mapper_registry.map_imperatively(
Publication, self.tables.publication
)
self.mapper_registry.map_imperatively(
Issue,
self.tables.issue,
properties={
"publication": relationship(
Publication,
backref=backref("issues", cascade="all, delete-orphan"),
)
},
)
self.mapper_registry.map_imperatively(
LocationName, self.tables.location_name
)
self.mapper_registry.map_imperatively(
Location,
self.tables.location,
properties={
"issue": relationship(
Issue,
backref=backref(
"locations",
lazy="joined",
cascade="all, delete-orphan",
),
),
"name": relationship(LocationName),
},
)
self.mapper_registry.map_imperatively(PageSize, self.tables.page_size)
self.mapper_registry.map_imperatively(
Magazine,
self.tables.magazine,
properties={
"location": relationship(
Location, backref=backref("magazine", uselist=False)
),
"size": relationship(PageSize),
},
)
if use_unions:
page_join = polymorphic_union(
{
"m": self.tables.page.join(self.tables.magazine_page),
"c": self.tables.page.join(self.tables.magazine_page).join(
self.tables.classified_page
),
"p": self.tables.page.select()
.where(self.tables.page.c.type == "p")
.subquery(),
},
None,
"page_join",
)
page_mapper = self.mapper_registry.map_imperatively(
Page,
self.tables.page,
with_polymorphic=("*", page_join),
polymorphic_on=page_join.c.type,
polymorphic_identity="p",
)
elif use_joins:
page_join = self.tables.page.outerjoin(
self.tables.magazine_page
).outerjoin(self.tables.classified_page)
page_mapper = self.mapper_registry.map_imperatively(
Page,
self.tables.page,
with_polymorphic=("*", page_join),
polymorphic_on=self.tables.page.c.type,
polymorphic_identity="p",
)
else:
page_mapper = self.mapper_registry.map_imperatively(
Page,
self.tables.page,
polymorphic_on=self.tables.page.c.type,
polymorphic_identity="p",
)
if use_unions:
magazine_join = polymorphic_union(
{
"m": self.tables.page.join(self.tables.magazine_page),
"c": self.tables.page.join(self.tables.magazine_page).join(
self.tables.classified_page
),
},
None,
"page_join",
)
magazine_page_mapper = self.mapper_registry.map_imperatively(
MagazinePage,
self.tables.magazine_page,
with_polymorphic=("*", magazine_join),
inherits=page_mapper,
polymorphic_identity="m",
properties={
"magazine": relationship(
Magazine,
backref=backref(
"pages", order_by=magazine_join.c.page_no
),
)
},
)
elif use_joins:
magazine_join = self.tables.page.join(
self.tables.magazine_page
).outerjoin(self.tables.classified_page)
magazine_page_mapper = self.mapper_registry.map_imperatively(
MagazinePage,
self.tables.magazine_page,
with_polymorphic=("*", magazine_join),
inherits=page_mapper,
polymorphic_identity="m",
properties={
"magazine": relationship(
Magazine,
backref=backref(
"pages", order_by=self.tables.page.c.page_no
),
)
},
)
else:
magazine_page_mapper = self.mapper_registry.map_imperatively(
MagazinePage,
self.tables.magazine_page,
inherits=page_mapper,
polymorphic_identity="m",
properties={
"magazine": relationship(
Magazine,
backref=backref(
"pages", order_by=self.tables.page.c.page_no
),
)
},
)
self.mapper_registry.map_imperatively(
ClassifiedPage,
self.tables.classified_page,
inherits=magazine_page_mapper,
polymorphic_identity="c",
primary_key=[self.tables.page.c.id],
)
@testing.combinations(
("unions", True, False),
("joins", False, True),
("plain", False, False),
id_="iaa",
)
def test_magazine_round_trip(self, use_unions, use_joins):
self._setup_mapping(use_unions, use_joins)
Publication = self.classes.Publication
session = fixture_session()
pub = self._generate_data()
session.add(pub)
session.commit()
session.close()
p = session.query(Publication).filter(Publication.name == "Test").one()
test_pub = self._generate_data()
eq_(p, test_pub)
eq_(
p.issues[0].locations[0].magazine.pages,
test_pub.issues[0].locations[0].magazine.pages,
)