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
|
"""Generic mapping to Select statements"""
from sqlalchemy.testing import assert_raises, assert_raises_message
import sqlalchemy as sa
from sqlalchemy import testing
from sqlalchemy import String, Integer, select, column
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import mapper, Session
from sqlalchemy.testing import eq_, AssertsCompiledSQL
from sqlalchemy.testing import fixtures
# 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()
mapper(Subset, selectable, primary_key=[selectable.c.x])
self.assert_compile(
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",
mapper,
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.InvalidRequestError, mapper, 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()
subset_mapper = mapper(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).mapped_table
eq_(
sess.query(Subset).filter(subset_select.c.data == 1).one(),
Subset(data=1),
)
|
