if self._limit_clause is not None or self._offset_clause is not None:
raise sa_exc.InvalidRequestError(
"Query.%s() being called on a Query which already has LIMIT "
- "or OFFSET applied. To modify the row-limited results of a "
- " Query, call from_self() first. "
- "Otherwise, call %s() before limit() or offset() "
+ "or OFFSET applied. Call %s() before limit() or offset() "
"are applied." % (meth, meth)
)
metadata.drop_all(self.engine)
@testing.requires.savepoints
- @testing.provide_metadata
def test_savepoints(self):
- metadata = self.metadata
+ metadata = MetaData()
some_table = Table(
"t",
mapper(SomeClass, some_table)
- metadata.create_all()
-
- session = Session(testing.db)
-
- target_strings = (
- session.connection().dialect.identifier_preparer._strings
- )
+ metadata.create_all(self.engine)
- session.close()
+ with Session(self.engine) as session:
+ target_strings = (
+ session.connection().dialect.identifier_preparer._strings
+ )
@profile_memory(
assert_no_sessions=False,
get_num_objects=lambda: len(target_strings),
)
def go():
- session = Session(testing.db)
- with session.transaction:
-
+ with Session(self.engine) as session, session.begin():
sc = SomeClass()
session.add(sc)
with session.begin_nested():
session.query(SomeClass).first()
- go()
+ try:
+ go()
+ finally:
+ metadata.drop_all(self.engine)
@testing.crashes("mysql+cymysql", "blocking")
def test_unicode_warnings(self):
session = fixture_session()
eq_(
- session.query(Joined).limit(10).offset(0).one(),
+ session.query(Joined)
+ .order_by(Joined.id)
+ .limit(10)
+ .offset(0)
+ .one(),
Joined(id=1, title="task 1", props_cnt=0),
)