--- /dev/null
+.. change::
+ :tags: change, orm
+ :tickets: 4412
+
+ Added a new function :func:`.close_all_sessions` which takes
+ over the task of the :meth:`.Session.close_all` method, which
+ is now deprecated as this is confusing as a classmethod.
+ Pull request courtesy Augustin Trancart.
Session Utilities
-----------------
+.. autofunction:: close_all_sessions
+
.. autofunction:: make_transient
.. autofunction:: make_transient_to_detached
from .relationships import RelationshipProperty # noqa
from .relationships import remote # noqa
from .scoping import scoped_session # noqa
+from .session import close_all_sessions # noqa
from .session import make_transient # noqa
from .session import make_transient_to_detached # noqa
from .session import object_session # noqa
"""Class-level methods for :class:`.Session`, :class:`.sessionmaker`."""
@classmethod
+ @util.deprecated(
+ "1.3",
+ "The :meth:`.Session.close_all` method is deprecated and will be "
+ "removed in a future release. Please refer to "
+ ":func:`.session.close_all_sessions`.",
+ )
def close_all(cls):
"""Close *all* sessions in memory."""
- for sess in _sessions.values():
- sess.close()
+ close_all_sessions()
@classmethod
@util.dependencies("sqlalchemy.orm.util")
)
+def close_all_sessions():
+ """Close all sessions in memory.
+
+ This function consults a global registry of all :class:`.Session` objects
+ and calls :meth:`.Session.close` on them, which resets them to a clean
+ state.
+
+ This function is not for general use but may be useful for test suites
+ within the teardown scheme.
+
+ .. versionadded:: 1.3
+
+ """
+
+ for sess in _sessions.values():
+ sess.close()
+
+
def make_transient(instance):
"""Alter the state of the given instance so that it is :term:`transient`.
class _ORMTest(object):
@classmethod
def teardown_class(cls):
- sa.orm.session.Session.close_all()
+ sa.orm.session.close_all_sessions()
sa.orm.clear_mappers()
self._setup_each_inserts()
def teardown(self):
- sa.orm.session.Session.close_all()
+ sa.orm.session.close_all_sessions()
self._teardown_each_mappers()
self._teardown_each_classes()
self._teardown_each_tables()
from sqlalchemy.orm import backref
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import clear_mappers
+from sqlalchemy.orm import close_all_sessions
from sqlalchemy.orm import column_property
from sqlalchemy.orm import composite
from sqlalchemy.orm import configure_mappers
Base = decl.declarative_base(testing.db)
def teardown(self):
- Session.close_all()
+ close_all_sessions()
clear_mappers()
Base.metadata.drop_all()
from sqlalchemy.ext.declarative import has_inherited_table
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import clear_mappers
+from sqlalchemy.orm import close_all_sessions
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import create_session
from sqlalchemy.orm import deferred
Base = decl.declarative_base(testing.db)
def teardown(self):
- Session.close_all()
+ close_all_sessions()
clear_mappers()
Base.metadata.drop_all()
from sqlalchemy.orm import base as orm_base
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import clear_mappers
+from sqlalchemy.orm import close_all_sessions
from sqlalchemy.orm import column_property
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import create_session
Base = decl.declarative_base(testing.db)
def teardown(self):
- Session.close_all()
+ close_all_sessions()
clear_mappers()
Base.metadata.drop_all()
from sqlalchemy import text
from sqlalchemy.orm import aliased
from sqlalchemy.orm import backref
+from sqlalchemy.orm import close_all_sessions
from sqlalchemy.orm import column_property
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import create_session
session.add_all([rscott, alien, brunner])
session.commit()
- session.close_all()
+ close_all_sessions()
self.d = session.query(Director).options(joinedload("*")).first()
assert len(list(session)) == 3
from sqlalchemy import testing
from sqlalchemy.orm import attributes
from sqlalchemy.orm import backref
+from sqlalchemy.orm import close_all_sessions
from sqlalchemy.orm import create_session
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import joinedload
class SessionUtilTest(_fixtures.FixtureTest):
run_inserts = None
+ def test_close_all_sessions(self):
+ users, User = self.tables.users, self.classes.User
+
+ mapper(User, users)
+
+ s1 = Session()
+ u1 = User()
+ s1.add(u1)
+
+ s2 = Session()
+ u2 = User()
+ s2.add(u2)
+
+ assert u1 in s1
+ assert u2 in s2
+
+ close_all_sessions()
+
+ assert u1 not in s1
+ assert u2 not in s2
+
+ def test_session_close_all_deprecated(self):
+ users, User = self.tables.users, self.classes.User
+
+ mapper(User, users)
+
+ s1 = Session()
+ u1 = User()
+ s1.add(u1)
+
+ s2 = Session()
+ u2 = User()
+ s2.add(u2)
+
+ assert u1 in s1
+ assert u2 in s2
+
+ with assertions.expect_deprecated(
+ r"The Session.close_all\(\) method is deprecated and will "
+ "be removed in a future release. "):
+ Session.close_all()
+
+ assert u1 not in s1
+ assert u2 not in s2
+
def test_object_session_raises(self):
User = self.classes.User
from sqlalchemy import testing
from sqlalchemy.orm import aliased
from sqlalchemy.orm import clear_mappers
+from sqlalchemy.orm import close_all_sessions
from sqlalchemy.orm import create_session
from sqlalchemy.orm import deferred
from sqlalchemy.orm import joinedload
session.add_all([rscott, alien, brunner])
session.commit()
- session.close_all()
+ close_all_sessions()
d = session.query(Director).options(subqueryload("*")).first()
assert len(list(session)) == 3
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
