1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
"""Generic mapping to Select statements"""
import sqlalchemy as sa
from sqlalchemy import column
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.orm import Session
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
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
# TODO: more tests mapping to selects
class SelectableNoFromsTest(fixtures.MappedTest, AssertsCompiledSQL):
@classmethod
def define_tables(cls, metadata):
Table(
"common",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("data", Integer),
Column("extra", String(45)),
)
@classmethod
def setup_classes(cls):
class Subset(cls.Comparable):
pass
def test_no_tables(self):
Subset = self.classes.Subset
selectable = select(column("x"), column("y"), column("z")).alias()
self.mapper_registry.map_imperatively(
Subset, selectable, primary_key=[selectable.c.x]
)
self.assert_compile(
fixture_session().query(Subset),
"SELECT anon_1.x AS anon_1_x, anon_1.y AS anon_1_y, "
"anon_1.z AS anon_1_z FROM (SELECT x, y, z) AS anon_1",
use_default_dialect=True,
)
def test_no_table_needs_pl(self):
Subset = self.classes.Subset
selectable = select(column("x"), column("y"), column("z")).alias()
assert_raises_message(
sa.exc.ArgumentError,
"could not assemble any primary key columns",
self.mapper_registry.map_imperatively,
Subset,
selectable,
)
def test_no_selects(self):
Subset, common = self.classes.Subset, self.tables.common
subset_select = select(common.c.id, common.c.data)
assert_raises(
sa.exc.ArgumentError,
self.mapper_registry.map_imperatively,
Subset,
subset_select,
)
def test_basic(self):
Subset, common = self.classes.Subset, self.tables.common
subset_select = select(common.c.id, common.c.data).alias()
self.mapper_registry.map_imperatively(Subset, subset_select)
sess = Session(bind=testing.db)
sess.add(Subset(data=1))
sess.flush()
sess.expunge_all()
eq_(sess.query(Subset).all(), [Subset(data=1)])
eq_(sess.query(Subset).filter(Subset.data == 1).one(), Subset(data=1))
eq_(sess.query(Subset).filter(Subset.data != 1).first(), None)
subset_select = sa.orm.class_mapper(Subset).persist_selectable
eq_(
sess.query(Subset).filter(subset_select.c.data == 1).one(),
Subset(data=1),
)
|
