def _setup_once_inserts(cls):
if cls.run_inserts == "once":
cls._load_fixtures()
- cls.insert_data()
+ with cls.bind.begin() as conn:
+ cls.insert_data(conn)
@classmethod
def _setup_once_tables(cls):
def _setup_each_inserts(self):
if self.run_inserts == "each":
self._load_fixtures()
- self.insert_data()
+ with self.bind.begin() as conn:
+ self.insert_data(conn)
def _teardown_each_tables(self):
if self.run_define_tables == "each":
return {}
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
pass
def sql_count_(self, count, fn):
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "data": "d1", "parent_id": None},
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.plain_pk.insert(),
[
{"id": 1, "data": "d1"},
],
)
- config.db.execute(
+ connection.execute(
cls.tables.has_dates.insert(),
[{"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)}],
)
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "data": "collate data1"},
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "x": 1, "y": 2, "q": "q1", "p": "p3"},
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "x": 1, "y": 2},
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "x": 1, "y": 2},
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "x": 1, "y": 2, "z": "z1"},
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.some_table.insert(),
[
{"id": 1, "data": "abcdefg"},
)
@classmethod
- def insert_data(cls):
- with config.db.begin() as conn:
- conn.execute(
- cls.tables.square.insert(),
- [{"id": 1, "side": 10}, {"id": 10, "side": 42}],
- )
+ def insert_data(cls, connection):
+ connection.execute(
+ cls.tables.square.insert(),
+ [{"id": 1, "side": 10}, {"id": 10, "side": 42}],
+ )
def test_select_all(self):
with config.db.connect() as conn:
)
@classmethod
- def insert_data(cls):
- config.db.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.plain_pk.insert(),
[
{"id": 1, "data": "d1"},
mapper(Child, child)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
parent, child = cls.tables.parent, cls.tables.child
- parent.insert().execute({"id": 1, "data": "p1"})
- child.insert().execute({"id": 1, "data": "p1c1", "parent_id": 1})
+ connection.execute(parent.insert(), {"id": 1, "data": "p1"})
+ connection.execute(
+ child.insert(), {"id": 1, "data": "p1c1", "parent_id": 1}
+ )
def test_merge_no_load(self):
Parent = self.classes.Parent
mapper(Child, child)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
parent, child = cls.tables.parent, cls.tables.child
- child.insert().execute(
- [{"id": i, "data": "c%d" % i} for i in range(1, 251)]
+ connection.execute(
+ child.insert(),
+ [{"id": i, "data": "c%d" % i} for i in range(1, 251)],
)
- parent.insert().execute(
+ connection.execute(
+ parent.insert(),
[
{
"id": i,
"child_id": (i % 250) + 1,
}
for i in range(1, 1000)
- ]
+ ],
)
def test_many_to_one_load_no_identity(self):
mapper(D, d)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
- s = Session()
+ s = Session(connection)
s.add_all(
[
A(
mapper(A, a)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A = cls.classes.A
- s = Session()
+ s = Session(connection)
s.add_all(
[
A(
mapper(C, c)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C = cls.classes("A", "B", "C")
- s = Session()
+ s = Session(connection)
s.add(A(bs=[B(cs=[C()]), B(cs=[C()])]))
s.commit()
mapper(G, g)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C, D, E, F, G = cls.classes("A", "B", "C", "D", "E", "F", "G")
- s = Session()
+ s = Session(connection)
s.add(
A(
bs=[B(cs=[C(ds=[D()])]), B(cs=[C()])],
mapper(A, a)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A = cls.classes.A
- s = Session()
+ s = Session(connection)
s.add_all([A(data="asdf") for i in range(5)])
s.commit()
__table_args__ = {"mysql_engine": "InnoDB"}
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A = cls.classes.A
B = cls.classes.B
# all the x/y are < 10
- s = Session()
+ s = Session(connection)
s.add_all(
[
A(x=5, y=5, bs=[B(x=4, y=4), B(x=2, y=8), B(x=7, y=1)]),
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
stuff = cls.tables.stuff
- testing.db.execute(
+ connection.execute(
stuff.insert(),
[
{"id": 1, "value": 1},
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
cls.data = data = [
dict(
id=i,
for i in range(1, 20)
]
- testing.db.execute(cls.tables.z_test.insert(), data)
+ connection.execute(cls.tables.z_test.insert(), data)
binary_table = cls.tables.binary_table
fname = os.path.join(
cls.stream = stream = file_.read(12000)
for i in range(1, 11):
- binary_table.insert().execute(id=i, data=stream)
+ connection.execute(binary_table.insert(), id=i, data=stream)
def test_lobs_without_convert(self):
engine = testing_engine(options=dict(auto_convert_lobs=False))
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
# TODO: why does setting hours to anything
# not affect the TZ in the DB col ?
class TZ(datetime.tzinfo):
def utcoffset(self, dt):
return datetime.timedelta(hours=4)
- cls.bind.execute(
+ connection.execute(
cls.tables.t.insert(),
{
"dtme": datetime.datetime(2012, 5, 10, 12, 15, 25),
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
data_table = cls.tables.data_table
- data_table.insert().execute(
- {"data": 3},
- {"data": 5},
- {"data": 7},
- {"data": 2},
- {"data": 15},
- {"data": 12},
- {"data": 6},
- {"data": 478},
- {"data": 52},
- {"data": 9},
+ connection.execute(
+ data_table.insert().values(
+ [
+ {"data": 3},
+ {"data": 5},
+ {"data": 7},
+ {"data": 2},
+ {"data": 15},
+ {"data": 12},
+ {"data": 6},
+ {"data": 478},
+ {"data": 52},
+ {"data": 9},
+ ]
+ )
)
@testing.fails_on(
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
params = [
dict(list(zip(("id", "name"), column_values)))
for column_values in [
]
]
- User.__table__.insert().execute(params)
- Address.__table__.insert().execute(
+ connection.execute(User.__table__.insert(), params)
+ connection.execute(
+ Address.__table__.insert(),
[
dict(list(zip(("id", "user_id", "email"), column_values)))
for column_values in [
(4, 8, "ed@lala.com"),
(5, 9, "fred@fred.com"),
]
- ]
+ ],
)
def test_aliased_join(self):
from sqlalchemy.orm import mapper
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
-from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm.collections import collection
from sqlalchemy.testing import assert_raises
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
UserKeyword, User, Keyword, Singular = (
cls.classes.UserKeyword,
cls.classes.User,
cls.classes.Singular,
)
- session = sessionmaker()()
+ session = Session(connection)
words = ("quick", "brown", "fox", "jumped", "over", "the", "lazy")
for ii in range(16):
user = User("user%d" % ii)
session.commit()
cls.u = user
cls.kw = user.keywords[0]
- cls.session = session
+
+ # TODO: this is not the correct pattern, use session per test
+ cls.session = Session(testing.db)
def _equivalent(self, q_proxy, q_direct):
proxy_sql = q_proxy.statement.compile(dialect=default.DefaultDialect())
data = Column(String(50))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- s = Session(testing.db)
+ s = Session(connection)
s.add_all(
[
A(id=1, bs=[B(data="b1"), B(data="b2")]),
deferred_data = deferred(Column(String(30)))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A = cls.classes.A
- s = Session()
+ s = Session(connection)
s.add(A(data="d1", deferred_data="d2"))
s.commit()
return self.fname
@classmethod
- def insert_data(cls):
- s = Session()
+ def insert_data(cls, connection):
+ s = Session(connection)
jill = cls.classes.Person(id=3, first_name="jill")
s.add(jill)
s.commit()
configure_mappers()
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
params = [
dict(list(zip(("id", "name"), column_values)))
for column_values in [
(10, "chuck"),
]
]
- users.insert().execute(params)
- addresses.insert().execute(
+ connection.execute(users.insert(), params)
+ connection.execute(
+ addresses.insert(),
[
dict(list(zip(("id", "user_id", "email"), column_values)))
for column_values in [
(4, 8, "ed@lala.com"),
(5, 9, "fred@fred.com"),
]
- ]
+ ],
)
def test_tables(self):
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
cls.e1 = e1 = Engineer(
name="dilbert",
cls.c2 = c2 = Company(name="Elbonia, Inc.")
c2.employees = [e3]
- sess = create_session()
+ sess = create_session(connection)
sess.add(c1)
sess.add(c2)
sess.flush()
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Status, Person, Engineer, Manager, Car = cls.classes(
"Status", "Person", "Engineer", "Manager", "Car"
)
- session = create_session()
+ session = create_session(connection)
active = Status(name="active")
dead = Status(name="dead")
__mapper_args__ = {"polymorphic_identity": "b"}
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A = cls.classes.A
- session = Session(testing.db)
+ session = Session(connection)
session.add_all(
[
A(id=1, discriminator="a"),
pass
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
person_insert = person_table.insert()
- person_insert.execute(id=1, name="alice")
- person_insert.execute(id=2, name="bob")
+ connection.execute(person_insert, dict(id=1, name="alice"))
+ connection.execute(person_insert, dict(id=2, name="bob"))
employee_insert = employee_table.insert()
- employee_insert.execute(id=2, salary=250, person_id=1) # alice
- employee_insert.execute(id=3, salary=200, person_id=2) # bob
+ connection.execute(
+ employee_insert, dict(id=2, salary=250, person_id=1)
+ ) # alice
+ connection.execute(
+ employee_insert, dict(id=3, salary=200, person_id=2)
+ ) # bob
def test_implicit(self):
person_mapper = mapper(Person, person_table)
)
@classmethod
- def insert_data(cls):
- refugees_table.insert().execute(
+ def insert_data(cls, connection):
+ connection.execute(
+ refugees_table.insert(),
dict(refugee_fid=1, name="refugee1"),
dict(refugee_fid=2, name="refugee2"),
)
- offices_table.insert().execute(
+ connection.execute(
+ offices_table.insert(),
dict(office_fid=1, name="office1"),
dict(office_fid=2, name="office2"),
)
a_sub_id = Column(ForeignKey("asub.id"))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, ASub, C = cls.classes("A", "B", "ASub", "C")
- s = Session()
+ s = Session(connection)
s.add(A(id=1, adata="adata", bs=[B(), B()]))
s.add(
ASub(
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Parent, ChildSubclass1, Other = cls.classes(
"Parent", "ChildSubclass1", "Other"
)
- session = Session()
+ session = Session(connection)
parent = Parent(id=1)
subclass1 = ChildSubclass1(id=1, parent=parent)
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
redefine_colprop = cls.redefine_colprop
include_base = cls.include_base
),
]
- session = Session()
+ session = Session(connection)
c = Company(name="company1")
c.employees = employees
session.add(c)
)
@classmethod
- def insert_data(cls):
- super(_PolymorphicTestBase, cls).insert_data()
+ def insert_data(cls, connection):
+ super(_PolymorphicTestBase, cls).insert_data(connection)
global all_employees, c1_employees, c2_employees
global c1, c2, e1, e2, e3, b1, m1
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Organization = cls.classes.Organization
e1 = Engineer(name="e1")
e2 = Engineer(name="e2")
e4 = Engineer(name="e4")
org1 = Organization(name="org1", engineers=[e1, e2])
org2 = Organization(name="org2", engineers=[e3, e4])
- sess = create_session()
+ sess = create_session(connection)
sess.add(org1)
sess.add(org2)
sess.flush()
mapper(Related, related)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
global p1, p2
Parent = cls.classes.Parent
Sub = cls.classes.Sub
Related = cls.classes.Related
- sess = Session()
+ sess = Session(connection)
r1, r2 = Related(data="r1"), Related(data="r2")
s1 = Sub(data="s1", related=r1)
s2 = Sub(data="s2", related=r2)
mapper(Sub, sub, inherits=Base, polymorphic_identity="s")
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
global p1, p2
Sub, Subparent = cls.classes.Sub, cls.classes.Subparent
- sess = create_session()
+ sess = create_session(connection)
p1 = Subparent(
data="p1",
children=[Sub(data="s1"), Sub(data="s2"), Sub(data="s3")],
mapper(D, cls.tables.d)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
B = cls.classes.B
C = cls.classes.C
D = cls.classes.D
- session = Session()
+ session = Session(connection)
d = D()
session.add_all([B(related=[d]), C(related=[d])])
mapper(D, cls.tables.d, inherits=A, polymorphic_identity="d")
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
B = cls.classes.B
- session = Session()
+ session = Session(connection)
session.add(B())
session.commit()
id = Column(Integer, primary_key=True)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
User, Fred, SubBar, Bar, SubFoo = cls.classes(
"User", "Fred", "SubBar", "Bar", "SubFoo"
)
sub_bar = SubBar(fred=fred)
rectangle = SubFoo(owner=user, baz=10, bar=bar, sub_bar=sub_bar)
- s = Session()
+ s = Session(connection)
s.add_all([user, fred, bar, sub_bar, rectangle])
s.commit()
child2 = relationship(Child2, viewonly=True)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Other, Child1 = cls.classes("Other", "Child1")
- s = Session()
+ s = Session(connection)
obj = Other(parent=Child1())
s.add(obj)
s.commit()
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Parent = cls.classes.Parent
SubChild1 = cls.classes.SubChild1
SubChild2 = cls.classes.SubChild2
- s = Session()
+ s = Session(connection)
s.add_all(
[
Parent(
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C = cls.classes("A", "B", "C")
- s = Session(testing.db)
+ s = Session(connection)
s.add_all([A(id=i) for i in range(1, 4)])
s.flush()
s.add_all(
A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C, D = cls.classes("A", "B", "C", "D")
- sess = Session()
+ sess = Session(connection)
for obj in [
B(id=1),
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Owner, Category, Option, Thing = (
cls.classes.Owner,
cls.classes.Category,
cls.classes.Thing,
)
- session = create_session()
+ session = create_session(connection)
o = Owner()
c = Category(name="Some Category")
mapper(cls.classes.Other, cls.tables.other)
@classmethod
- def insert_data(cls):
- sess = Session()
+ def insert_data(cls, connection):
+ sess = Session(connection)
sess.add_all(
[
cls.classes.Data(
mapper(Foo, foo)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Pref, User, Extra = (
cls.classes.Pref,
cls.classes.User,
u1 = User(name="ed", pref=Pref(data="pref 1", extra=[Extra()]))
u2 = User(name="jack", pref=Pref(data="pref 2", extra=[Extra()]))
u3 = User(name="foo", pref=Pref(data="pref 3", extra=[Extra()]))
- sess = create_session()
+ sess = create_session(connection)
sess.add_all((u1, u2, u3))
sess.flush()
sess.close()
name = sa.Column(sa.String(10))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Node = cls.classes.Node
- session = Session()
+ session = Session(connection)
session.add_all(
[
Node(id=1, name="name"),
b_expr = query_expression()
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- s = Session()
+ s = Session(connection)
s.add_all(
[
]
@classmethod
- def insert_data(cls):
- s = Session(testing.db)
+ def insert_data(cls, connection):
+ s = Session(connection)
s.add_all(cls._fixture_data())
s.commit()
]
@classmethod
- def insert_data(cls):
- s = Session(testing.db)
+ def insert_data(cls, connection):
+ s = Session(connection)
s.add_all(cls._fixture_data())
s.commit()
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
b_table, a_table = cls.tables.b_table, cls.tables.a_table
- a_table.insert().execute(dict(id=1), dict(id=2), dict(id=3))
- b_table.insert().execute(
- dict(id=1, parent_a_id=2, parent_b1_id=None, parent_b2_id=None),
- dict(id=2, parent_a_id=1, parent_b1_id=1, parent_b2_id=None),
- dict(id=3, parent_a_id=1, parent_b1_id=1, parent_b2_id=2),
- dict(id=4, parent_a_id=3, parent_b1_id=1, parent_b2_id=None),
- dict(id=5, parent_a_id=3, parent_b1_id=None, parent_b2_id=2),
- dict(id=6, parent_a_id=1, parent_b1_id=1, parent_b2_id=3),
- dict(id=7, parent_a_id=2, parent_b1_id=None, parent_b2_id=3),
- dict(id=8, parent_a_id=2, parent_b1_id=1, parent_b2_id=2),
- dict(id=9, parent_a_id=None, parent_b1_id=1, parent_b2_id=None),
- dict(id=10, parent_a_id=3, parent_b1_id=7, parent_b2_id=2),
- dict(id=11, parent_a_id=3, parent_b1_id=1, parent_b2_id=8),
- dict(id=12, parent_a_id=2, parent_b1_id=5, parent_b2_id=2),
- dict(id=13, parent_a_id=3, parent_b1_id=4, parent_b2_id=4),
- dict(id=14, parent_a_id=3, parent_b1_id=7, parent_b2_id=2),
+ connection.execute(
+ a_table.insert(), [dict(id=1), dict(id=2), dict(id=3)]
+ )
+ connection.execute(
+ b_table.insert(),
+ [
+ dict(
+ id=1, parent_a_id=2, parent_b1_id=None, parent_b2_id=None
+ ),
+ dict(id=2, parent_a_id=1, parent_b1_id=1, parent_b2_id=None),
+ dict(id=3, parent_a_id=1, parent_b1_id=1, parent_b2_id=2),
+ dict(id=4, parent_a_id=3, parent_b1_id=1, parent_b2_id=None),
+ dict(id=5, parent_a_id=3, parent_b1_id=None, parent_b2_id=2),
+ dict(id=6, parent_a_id=1, parent_b1_id=1, parent_b2_id=3),
+ dict(id=7, parent_a_id=2, parent_b1_id=None, parent_b2_id=3),
+ dict(id=8, parent_a_id=2, parent_b1_id=1, parent_b2_id=2),
+ dict(
+ id=9, parent_a_id=None, parent_b1_id=1, parent_b2_id=None
+ ),
+ dict(id=10, parent_a_id=3, parent_b1_id=7, parent_b2_id=2),
+ dict(id=11, parent_a_id=3, parent_b1_id=1, parent_b2_id=8),
+ dict(id=12, parent_a_id=2, parent_b1_id=5, parent_b2_id=2),
+ dict(id=13, parent_a_id=3, parent_b1_id=4, parent_b2_id=4),
+ dict(id=14, parent_a_id=3, parent_b1_id=7, parent_b2_id=2),
+ ],
)
def test_eager_load(self):
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
stuff, users = cls.tables.stuff, cls.tables.users
- users.insert().execute(
+ connection.execute(
+ users.insert(),
{"id": 1, "name": "user1"},
{"id": 2, "name": "user2"},
{"id": 3, "name": "user3"},
)
- stuff.insert().execute(
+ connection.execute(
+ stuff.insert(),
{"id": 1, "user_id": 1, "date": datetime.date(2007, 10, 15)},
{"id": 2, "user_id": 1, "date": datetime.date(2007, 12, 15)},
{"id": 3, "user_id": 1, "date": datetime.date(2007, 11, 15)},
b_id = Column(ForeignKey("b.id"))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C = cls.classes("A", "B", "C")
- s = Session()
+ s = Session(connection)
s.add(A(id=1, bs=[B(cs=[C()])]))
s.add(A(id=2))
s.commit()
a_id = Column(ForeignKey("a.id"))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- s = Session(testing.db)
+ s = Session(connection)
s.add(A(bs=[B(), B(), B()]))
s.commit()
pass
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
people, engineers = cls.tables.people, cls.tables.engineers
- people.insert().execute(
+ connection.execute(
+ people.insert(),
{"person_id": 1, "name": "person1", "type": "person"},
{"person_id": 2, "name": "engineer1", "type": "engineer"},
{"person_id": 3, "name": "engineer2", "type": "engineer"},
)
- engineers.insert().execute(
+ connection.execute(
+ engineers.insert(),
{"person_id": 2, "status": "new engineer"},
{"person_id": 3, "status": "old engineer"},
)
mapper(D, d, inherits=A, polymorphic_identity="d")
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, C, B = (cls.classes.A, cls.classes.C, cls.classes.B)
- sess = create_session()
+ sess = create_session(connection)
sess.add_all(
[
B(name="b1"),
mapper(cls.classes.Bar, cls.tables.foo_bar)
@classmethod
- def insert_data(cls):
- s = Session()
+ def insert_data(cls, connection):
+ s = Session(connection)
s.add_all([cls.classes.Foo(id=1, bar_id=2), cls.classes.Bar(id=3)])
s.commit()
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Node = cls.classes.Node
- sess = create_session()
+ sess = create_session(connection)
n1 = Node(data="n1")
n1.append(Node(data="n11"))
n1.append(Node(data="n12"))
pass
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Node, nodes, node_to_nodes = (
cls.classes.Node,
cls.tables.nodes,
)
},
)
- sess = create_session()
+ sess = create_session(connection)
n1 = Node(data="n1")
n2 = Node(data="n2")
n3 = Node(data="n3")
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
stuff, user_t = cls.tables.stuff, cls.tables.user_t
- user_t.insert().execute(
+ connection.execute(
+ user_t.insert(),
{"id": 1, "name": "user1"},
{"id": 2, "name": "user2"},
{"id": 3, "name": "user3"},
)
- stuff.insert().execute(
+ connection.execute(
+ stuff.insert(),
{"id": 1, "user_id": 1, "date": datetime.date(2007, 10, 15)},
{"id": 2, "user_id": 1, "date": datetime.date(2007, 12, 15)},
{"id": 3, "user_id": 1, "date": datetime.date(2007, 11, 15)},
mapper(Related, related)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
child1, child2, base, related = (
cls.tables.child1,
cls.tables.child2,
cls.tables.related,
)
- base.insert().execute(
+ connection.execute(
+ base.insert(),
[
{"id": 1, "type": "child1"},
{"id": 2, "type": "child1"},
{"id": 4, "type": "child2"},
{"id": 5, "type": "child2"},
{"id": 6, "type": "child2"},
- ]
+ ],
)
- child2.insert().execute([{"id": 4}, {"id": 5}, {"id": 6}])
- child1.insert().execute(
+ connection.execute(child2.insert(), [{"id": 4}, {"id": 5}, {"id": 6}])
+ connection.execute(
+ child1.insert(),
[
{"id": 1, "child2id": 4},
{"id": 2, "child2id": 5},
{"id": 3, "child2id": 6},
- ]
+ ],
)
- related.insert().execute(
- [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}, {"id": 6}]
+ connection.execute(
+ related.insert(),
+ [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}, {"id": 6}],
)
def test_contains_eager(self):
mapper(Thing, thing, properties={"name": deferred(thing.c.name)})
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
thing, human = cls.tables.thing, cls.tables.human
- thing.insert().execute([{"id": 1, "name": "Chair"}])
+ connection.execute(thing.insert(), [{"id": 1, "name": "Chair"}])
- human.insert().execute(
- [{"id": 1, "thing_id": 1, "name": "Clark Kent"}]
+ connection.execute(
+ human.insert(), [{"id": 1, "thing_id": 1, "name": "Clark Kent"}]
)
def _test(self, thing):
assert Address.user.property._use_get is False
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
User, Address = cls.classes.User, cls.classes.Address
- s = Session()
+ s = Session(connection)
s.add_all(
[
User(
name = Column(String(10))
@classmethod
- def insert_data(cls):
- s = Session(testing.db)
+ def insert_data(cls, connection):
+ s = Session(connection)
s.add_all(cls._fixture())
s.commit()
c = relationship("C", backref="ds")
@classmethod
- def insert_data(cls):
- s = Session(testing.db)
+ def insert_data(cls, connection):
+ s = Session(connection)
s.add_all(cls._fixture())
s.commit()
mapper(D, tbl_d, properties=dict(b_row=relationship(B)))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, C, B, D = (
cls.classes.A,
cls.classes.C,
cls.classes.D,
)
- session = create_session()
+ session = create_session(connection)
a = A(name="a1")
b = B(name="b1")
c = C(name="c1", a_row=a)
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Venue = cls.classes.Venue
- s = Session()
+ s = Session(connection)
s.add_all(
[
Venue(name="parent1"),
ip_addr = Column(Integer, primary_key=True)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Network, Address = cls.classes.Network, cls.classes.Address
- s = Session(testing.db)
+ s = Session(connection)
s.add_all(
[
mapper(D, d)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
- sess = Session()
+ sess = Session(connection)
a1, a2, a3, a4 = A(name="a1"), A(name="a2"), A(name="a3"), A(name="a4")
b1, b2, b3, b4 = B(name="b1"), B(name="b2"), B(name="b3"), B(name="b4")
c1, c2, c3, c4 = C(name="c1"), C(name="c2"), C(name="c3"), C(name="c4")
mapper(B, b)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- s = Session()
+ s = Session(connection)
s.add_all(
[
A(id=1, b_ids="1"),
mapper(Paperwork, paperwork)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
e1 = Engineer(primary_language="java")
e2 = Engineer(primary_language="c++")
Paperwork(description="tps report #2"),
]
e2.paperwork = [Paperwork(description="tps report #3")]
- sess = create_session()
+ sess = create_session(connection)
sess.add_all([e1, e2])
sess.flush()
name = Column(String(50))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Company, Programmer, Manager, GolfSwing, Language = cls.classes(
"Company", "Programmer", "Manager", "GolfSwing", "Language"
)
),
],
)
- sess = Session()
+ sess = Session(connection)
sess.add_all([c1, c2])
sess.commit()
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- session = Session()
+ session = Session(connection)
session.add_all(
[
A(id1=i, id2=i + 2, bs=[B(id=(i * 6) + j) for j in range(6)])
a = relationship("A", back_populates="bs")
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- session = Session()
+ session = Session(connection)
session.add_all(
[
A(id=i, bs=[B(id=(i * 6) + j) for j in range(1, 6)])
mapper(MachineType, machine_type)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
c1 = cls._fixture()
- sess = create_session()
+ sess = create_session(connection)
sess.add(c1)
sess.flush()
__mapper_args__ = {"polymorphic_identity": "bar"}
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Foo, Bar = cls.classes("Foo", "Bar")
- session = Session()
+ session = Session(connection)
target = Bar(id=1)
b1 = Bar(id=2, foo=Foo(id=3, foo=target))
session.add(b1)
id = Column(Integer, primary_key=True)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, A2, B, C1o2m, C2o2m, C1m2o, C2m2o = cls.classes(
"A", "A2", "B", "C1o2m", "C2o2m", "C1m2o", "C2m2o"
)
- s = Session()
+ s = Session(connection)
b = B(
c1_o2m=[C1o2m()], c2_o2m=[C2o2m()], c1_m2o=C1m2o(), c2_m2o=C2m2o()
user_id = Column(Integer, ForeignKey("user.id"))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
EmployerUser, Role = cls.classes("EmployerUser", "Role")
- s = Session()
+ s = Session(connection)
s.add(EmployerUser(roles=[Role(), Role(), Role()]))
s.commit()
y = Column(Integer)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- s = Session()
+ s = Session(connection)
b1, b2 = B(id=1, x=5, y=9), B(id=2, x=10, y=8)
s.add_all(
[
y = Column(Integer)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, B = cls.classes("A", "B")
- s = Session()
+ s = Session(connection)
b1, b2 = B(id=1, x=5, y=9), B(id=2, x=10, y=8)
s.add_all(
[
parent = relationship("ParentB", back_populates="children")
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
ParentA, ParentB, ChildA, ChildB = cls.classes(
"ParentA", "ParentB", "ChildA", "ChildB"
)
- session = Session()
+ session = Session(connection)
parent_a = ParentA(id=1)
parent_b = ParentB(id=2)
for i in range(10):
mapper(Page, pages)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
e1 = Engineer(primary_language="java")
e2 = Engineer(primary_language="c++")
),
]
e2.paperwork = [Paperwork(description="tps report #3")]
- sess = create_session()
+ sess = create_session(connection)
sess.add_all([e1, e2])
sess.flush()
mapper(MachineType, machine_type)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
c1 = cls._fixture()
- sess = create_session()
+ sess = create_session(connection)
sess.add(c1)
sess.flush()
movie_id = Column(Integer, ForeignKey("movie.id"))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Movie = cls.classes.Movie
Director = cls.classes.Director
DirectorPhoto = cls.classes.DirectorPhoto
Movie(title="Manhattan", credits=[Credit(), Credit()]),
Movie(title="Sweet and Lowdown", credits=[Credit()]),
]
- sess = create_session()
+ sess = create_session(connection)
sess.add_all([d])
sess.flush()
)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Parent = cls.classes.Parent
Child = cls.classes.Child
- s = Session()
+ s = Session(connection)
s.add(Parent(name="parent", children=[Child(name="c1")]))
s.commit()
__mapper_args__ = {"polymorphic_identity": "bar"}
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Foo, Bar = cls.classes("Foo", "Bar")
- session = Session()
+ session = Session(connection)
target = Bar(id=1)
b1 = Bar(id=2, foo=Foo(id=3, foo=target))
session.add(b1)
id = Column(Integer, primary_key=True)
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
A, A2, B, C1o2m, C2o2m, C1m2o, C2m2o = cls.classes(
"A", "A2", "B", "C1o2m", "C2o2m", "C1m2o", "C2m2o"
)
- s = Session()
+ s = Session(connection)
b = B(
c1_o2m=[C1o2m()], c2_o2m=[C2o2m()], c1_m2o=C1m2o(), c2_m2o=C2m2o()
pass
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
users = cls.tables.users
- users.insert().execute(
+ connection.execute(
+ users.insert(),
[
dict(id=1, name="john", age_int=25),
dict(id=2, name="jack", age_int=47),
dict(id=3, name="jill", age_int=29),
dict(id=4, name="jane", age_int=37),
- ]
+ ],
)
@classmethod
pass
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
users = cls.tables.users
- users.insert().execute(
+ connection.execute(
+ users.insert(),
[
dict(id=1, name="john", age=25),
dict(id=2, name="jack", age=47),
dict(id=3, name="jill", age=29),
dict(id=4, name="jane", age=37),
- ]
+ ],
)
documents = cls.tables.documents
- documents.insert().execute(
+ connection.execute(
+ documents.insert(),
[
dict(id=1, user_id=1, title="foo"),
dict(id=2, user_id=1, title="bar"),
dict(id=3, user_id=2, title="baz"),
- ]
+ ],
)
@classmethod
pass
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
users = cls.tables.users
- users.insert().execute(
- [dict(id=1), dict(id=2), dict(id=3), dict(id=4)]
+ connection.execute(
+ users.insert(), [dict(id=1), dict(id=2), dict(id=3), dict(id=4)]
)
documents = cls.tables.documents
- documents.insert().execute(
+ connection.execute(
+ documents.insert(),
[
dict(id=1, user_id=1, title="foo"),
dict(id=2, user_id=1, title="bar"),
dict(id=4, user_id=2, title="hoho"),
dict(id=5, user_id=3, title="lala"),
dict(id=6, user_id=3, title="bleh"),
- ]
+ ],
)
@classmethod
manager_name = Column(String(50))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
Engineer, Person, Manager = (
cls.classes.Engineer,
cls.classes.Person,
cls.classes.Manager,
)
- s = Session(testing.db)
+ s = Session(connection)
s.add_all(
[
Engineer(name="e1", engineer_name="e1"),
Table("data", metadata, Column("x", Integer), Column("y", Integer))
@classmethod
- def insert_data(cls):
+ def insert_data(cls, connection):
data = cls.tables.data
- with testing.db.connect() as conn:
- conn.execute(data.insert(), [{"x": 2, "y": 5}, {"x": 7, "y": 12}])
+ connection.execute(
+ data.insert(), [{"x": 2, "y": 5}, {"x": 7, "y": 12}]
+ )
@testing.provide_metadata
def test_insert_from_select_override_defaults(self, connection):
)
@classmethod
- def insert_data(cls):
- cls.tables.keyed1.insert().execute(dict(b="a1", q="c1"))
- cls.tables.keyed2.insert().execute(dict(a="a2", b="b2"))
- cls.tables.keyed3.insert().execute(dict(a="a3", d="d3"))
- cls.tables.keyed4.insert().execute(dict(b="b4", q="q4"))
- cls.tables.content.insert().execute(type="t1")
+ def insert_data(cls, connection):
+ conn = connection
+ conn.execute(cls.tables.keyed1.insert(), dict(b="a1", q="c1"))
+ conn.execute(cls.tables.keyed2.insert(), dict(a="a2", b="b2"))
+ conn.execute(cls.tables.keyed3.insert(), dict(a="a3", d="d3"))
+ conn.execute(cls.tables.keyed4.insert(), dict(b="b4", q="q4"))
+ conn.execute(cls.tables.content.insert(), dict(type="t1"))
if testing.requires.schemas.enabled:
- cls.tables[
- "%s.wschema" % testing.config.test_schema
- ].insert().execute(dict(b="a1", q="c1"))
+ conn.execute(
+ cls.tables["%s.wschema" % testing.config.test_schema].insert(),
+ dict(b="a1", q="c1"),
+ )
@testing.requires.schemas
def test_keyed_accessor_wschema(self):
)
@classmethod
- def insert_data(cls):
- cls.tables.text1.insert().execute(
- [dict(a="a1", b="b1", c="c1", d="d1")]
+ def insert_data(cls, connection):
+ connection.execute(
+ cls.tables.text1.insert(), [dict(a="a1", b="b1", c="c1", d="d1")],
)
def test_via_column(self):
)
@classmethod
- def insert_data(cls):
- cls.engine.execute(
+ def insert_data(cls, connection):
+ connection.execute(
cls.tables.test.insert(),
[{"x": i, "y": "t_%d" % i} for i in range(1, 12)],
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
