Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/orm/scoping.py : 50%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/scoping.py
2# Copyright (C) 2005-2020 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: http://www.opensource.org/licenses/mit-license.php
8from . import class_mapper
9from . import exc as orm_exc
10from .session import Session
11from .. import exc as sa_exc
12from ..util import ScopedRegistry
13from ..util import ThreadLocalRegistry
14from ..util import warn
17__all__ = ["scoped_session"]
20class scoped_session(object):
21 """Provides scoped management of :class:`.Session` objects.
23 See :ref:`unitofwork_contextual` for a tutorial.
25 """
27 session_factory = None
28 """The `session_factory` provided to `__init__` is stored in this
29 attribute and may be accessed at a later time. This can be useful when
30 a new non-scoped :class:`.Session` or :class:`_engine.Connection` to the
31 database is needed."""
33 def __init__(self, session_factory, scopefunc=None):
34 """Construct a new :class:`.scoped_session`.
36 :param session_factory: a factory to create new :class:`.Session`
37 instances. This is usually, but not necessarily, an instance
38 of :class:`.sessionmaker`.
39 :param scopefunc: optional function which defines
40 the current scope. If not passed, the :class:`.scoped_session`
41 object assumes "thread-local" scope, and will use
42 a Python ``threading.local()`` in order to maintain the current
43 :class:`.Session`. If passed, the function should return
44 a hashable token; this token will be used as the key in a
45 dictionary in order to store and retrieve the current
46 :class:`.Session`.
48 """
49 self.session_factory = session_factory
51 if scopefunc:
52 self.registry = ScopedRegistry(session_factory, scopefunc)
53 else:
54 self.registry = ThreadLocalRegistry(session_factory)
56 def __call__(self, **kw):
57 r"""Return the current :class:`.Session`, creating it
58 using the :attr:`.scoped_session.session_factory` if not present.
60 :param \**kw: Keyword arguments will be passed to the
61 :attr:`.scoped_session.session_factory` callable, if an existing
62 :class:`.Session` is not present. If the :class:`.Session` is present
63 and keyword arguments have been passed,
64 :exc:`~sqlalchemy.exc.InvalidRequestError` is raised.
66 """
67 if kw:
68 if self.registry.has():
69 raise sa_exc.InvalidRequestError(
70 "Scoped session is already present; "
71 "no new arguments may be specified."
72 )
73 else:
74 sess = self.session_factory(**kw)
75 self.registry.set(sess)
76 return sess
77 else:
78 return self.registry()
80 def remove(self):
81 """Dispose of the current :class:`.Session`, if present.
83 This will first call :meth:`.Session.close` method
84 on the current :class:`.Session`, which releases any existing
85 transactional/connection resources still being held; transactions
86 specifically are rolled back. The :class:`.Session` is then
87 discarded. Upon next usage within the same scope,
88 the :class:`.scoped_session` will produce a new
89 :class:`.Session` object.
91 """
93 if self.registry.has():
94 self.registry().close()
95 self.registry.clear()
97 def configure(self, **kwargs):
98 """reconfigure the :class:`.sessionmaker` used by this
99 :class:`.scoped_session`.
101 See :meth:`.sessionmaker.configure`.
103 """
105 if self.registry.has():
106 warn(
107 "At least one scoped session is already present. "
108 " configure() can not affect sessions that have "
109 "already been created."
110 )
112 self.session_factory.configure(**kwargs)
114 def query_property(self, query_cls=None):
115 """return a class property which produces a :class:`_query.Query`
116 object
117 against the class and the current :class:`.Session` when called.
119 e.g.::
121 Session = scoped_session(sessionmaker())
123 class MyClass(object):
124 query = Session.query_property()
126 # after mappers are defined
127 result = MyClass.query.filter(MyClass.name=='foo').all()
129 Produces instances of the session's configured query class by
130 default. To override and use a custom implementation, provide
131 a ``query_cls`` callable. The callable will be invoked with
132 the class's mapper as a positional argument and a session
133 keyword argument.
135 There is no limit to the number of query properties placed on
136 a class.
138 """
140 class query(object):
141 def __get__(s, instance, owner):
142 try:
143 mapper = class_mapper(owner)
144 if mapper:
145 if query_cls:
146 # custom query class
147 return query_cls(mapper, session=self.registry())
148 else:
149 # session's configured query class
150 return self.registry().query(mapper)
151 except orm_exc.UnmappedClassError:
152 return None
154 return query()
157ScopedSession = scoped_session
158"""Old name for backwards compatibility."""
161def instrument(name):
162 def do(self, *args, **kwargs):
163 return getattr(self.registry(), name)(*args, **kwargs)
165 return do
168for meth in Session.public_methods:
169 setattr(scoped_session, meth, instrument(meth))
172def makeprop(name):
173 def set_(self, attr):
174 setattr(self.registry(), name, attr)
176 def get(self):
177 return getattr(self.registry(), name)
179 return property(get, set_)
182for prop in (
183 "bind",
184 "dirty",
185 "deleted",
186 "new",
187 "identity_map",
188 "is_active",
189 "autoflush",
190 "no_autoflush",
191 "info",
192 "autocommit",
193):
194 setattr(scoped_session, prop, makeprop(prop))
197def clslevel(name):
198 def do(cls, *args, **kwargs):
199 return getattr(Session, name)(*args, **kwargs)
201 return classmethod(do)
204for prop in ("close_all", "object_session", "identity_key"):
205 setattr(scoped_session, prop, clslevel(prop))