Method test_scalar_proxy

test/ext/test_associationproxy.py:1047–1164

(self)

class ScalarTest(fixtures.MappedTest):
    @testing.provide_metadata
    def test_scalar_proxy(self):
        metadata = self.metadata

        parents_table = Table(
            class="st">"Parent",
            metadata,
            Column(
                class="st">"id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column(class="st">"name", String(128)),
        )
        children_table = Table(
            class="st">"Children",
            metadata,
            Column(
                class="st">"id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column(class="st">"parent_id", Integer, ForeignKey(class="st">"Parent.id")),
            Column(class="st">"foo", String(128)),
            Column(class="st">"bar", String(128)),
            Column(class="st">"baz", String(128)),
        )

        class Parent:
            foo = association_proxy(class="st">"child", class="st">"foo")
            bar = association_proxy(
                class="st">"child", class="st">"bar", creator=lambda v: Child(bar=v)
            )
            baz = association_proxy(
                class="st">"child", class="st">"baz", creator=lambda v: Child(baz=v)
            )

            def __init__(self, name):
                self.name = name

        class Child:
            def __init__(self, **kw):
                for attr in kw:
                    setattr(self, attr, kw[attr])

        self.mapper_registry.map_imperatively(
            Parent,
            parents_table,
            properties={
                class="st">"child": relationship(
                    Child, lazy=class="st">"joined", backref=class="st">"parent", uselist=False
                )
            },
        )
        self.mapper_registry.map_imperatively(Child, children_table)

        metadata.create_all(testing.db)
        session = fixture_session()

        def roundtrip(obj):
            if obj not in session:
                session.add(obj)
            session.flush()