codekingpro/portable-devtools
115k
1# ext/asyncio/scoping.py
2# Copyright (C) 2005-2026 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: https://www.opensource.org/licenses/mit-license.php
7
8from __future__ import annotations
9
10from typing import Any
11from typing import Callable
12from typing import Generic
13from typing import Iterable
14from typing import Iterator
15from typing import Optional
16from typing import overload
17from typing import Sequence
18from typing import Tuple
19from typing import Type
20from typing import TYPE_CHECKING
21from typing import TypeVar
22from typing import Union
23
24from .session import _AS
25from .session import async_sessionmaker
26from .session import AsyncSession
27from ... import exc as sa_exc
28from ... import util
29from ...orm.session import Session
30from ...util import create_proxy_methods
31from ...util import ScopedRegistry
32from ...util import warn
33from ...util import warn_deprecated
34
35if TYPE_CHECKING:
36 from .engine import AsyncConnection
37 from .result import AsyncResult
38 from .result import AsyncScalarResult
39 from .session import AsyncSessionTransaction
40 from ...engine import Connection
41 from ...engine import Engine
42 from ...engine import Result
43 from ...engine import Row
44 from ...engine import RowMapping
45 from ...engine.interfaces import _CoreAnyExecuteParams
46 from ...engine.interfaces import CoreExecuteOptionsParameter
47 from ...engine.result import ScalarResult
48 from ...orm._typing import _IdentityKeyType
49 from ...orm._typing import _O
50 from ...orm._typing import OrmExecuteOptionsParameter
51 from ...orm.interfaces import ORMOption
52 from ...orm.session import _BindArguments
53 from ...orm.session import _EntityBindKey
54 from ...orm.session import _PKIdentityArgument
55 from ...orm.session import _SessionBind
56 from ...sql.base import Executable
57 from ...sql.elements import ClauseElement
58 from ...sql.selectable import ForUpdateParameter
59 from ...sql.selectable import TypedReturnsRows
60
61_T = TypeVar("_T", bound=Any)
62
63
64@create_proxy_methods(
65 AsyncSession,
66 ":class:`_asyncio.AsyncSession`",
67 ":class:`_asyncio.scoping.async_scoped_session`",
68 classmethods=["close_all", "object_session", "identity_key"],
69 methods=[
70 "__contains__",
71 "__iter__",
72 "aclose",
73 "add",
74 "add_all",
75 "begin",
76 "begin_nested",
77 "close",
78 "reset",
79 "commit",
80 "connection",
81 "delete",
82 "execute",
83 "expire",
84 "expire_all",
85 "expunge",
86 "expunge_all",
87 "flush",
88 "get_bind",
89 "is_modified",
90 "invalidate",
91 "merge",
92 "refresh",
93 "rollback",
94 "scalar",
95 "scalars",
96 "get",
97 "get_one",
98 "stream",
99 "stream_scalars",
100 ],
101 attributes=[
102 "bind",
103 "dirty",
104 "deleted",
105 "new",
106 "identity_map",
107 "is_active",
108 "autoflush",
109 "no_autoflush",
110 "info",
111 ],
112 use_intermediate_variable=["get"],
113)
114class async_scoped_session(Generic[_AS]):
115 """Provides scoped management of :class:`.AsyncSession` objects.
116
117 See the section :ref:`asyncio_scoped_session` for usage details.
118
119 .. versionadded:: 1.4.19
120
121
122 """
123
124 _support_async = True
125
126 session_factory: async_sessionmaker[_AS]
127 """The `session_factory` provided to `__init__` is stored in this
128 attribute and may be accessed at a later time. This can be useful when
129 a new non-scoped :class:`.AsyncSession` is needed."""
130
131 registry: ScopedRegistry[_AS]
132
133 def __init__(
134 self,
135 session_factory: async_sessionmaker[_AS],
136 scopefunc: Callable[[], Any],
137 ):
138 """Construct a new :class:`_asyncio.async_scoped_session`.
139
140 :param session_factory: a factory to create new :class:`_asyncio.AsyncSession`
141 instances. This is usually, but not necessarily, an instance
142 of :class:`_asyncio.async_sessionmaker`.
143
144 :param scopefunc: function which defines
145 the current scope. A function such as ``asyncio.current_task``
146 may be useful here.
147
148 """ # noqa: E501
149
150 self.session_factory = session_factory
151 self.registry = ScopedRegistry(session_factory, scopefunc)
152
153 @property
154 def _proxied(self) -> _AS:
155 return self.registry()
156
157 def __call__(self, **kw: Any) -> _AS:
158 r"""Return the current :class:`.AsyncSession`, creating it
159 using the :attr:`.scoped_session.session_factory` if not present.
160
161 :param \**kw: Keyword arguments will be passed to the
162 :attr:`.scoped_session.session_factory` callable, if an existing
163 :class:`.AsyncSession` is not present. If the
164 :class:`.AsyncSession` is present
165 and keyword arguments have been passed,
166 :exc:`~sqlalchemy.exc.InvalidRequestError` is raised.
167
168 """
169 if kw:
170 if self.registry.has():
171 raise sa_exc.InvalidRequestError(
172 "Scoped session is already present; "
173 "no new arguments may be specified."
174 )
175 else:
176 sess = self.session_factory(**kw)
177 self.registry.set(sess)
178 else:
179 sess = self.registry()
180 if not self._support_async and sess._is_asyncio:
181 warn_deprecated(
182 "Using `scoped_session` with asyncio is deprecated and "
183 "will raise an error in a future version. "
184 "Please use `async_scoped_session` instead.",
185 "1.4.23",
186 )
187 return sess
188
189 def configure(self, **kwargs: Any) -> None:
190 """reconfigure the :class:`.sessionmaker` used by this
191 :class:`.scoped_session`.
192
193 See :meth:`.sessionmaker.configure`.
194
195 """
196
197 if self.registry.has():
198 warn(
199 "At least one scoped session is already present. "
200 " configure() can not affect sessions that have "
201 "already been created."
202 )
203
204 self.session_factory.configure(**kwargs)
205
206 async def remove(self) -> None:
207 """Dispose of the current :class:`.AsyncSession`, if present.
208
209 Different from scoped_session's remove method, this method would use
210 await to wait for the close method of AsyncSession.
211
212 """
213
214 if self.registry.has():
215 await self.registry().close()
216 self.registry.clear()
217
218 # START PROXY METHODS async_scoped_session
219
220 # code within this block is **programmatically,
221 # statically generated** by tools/generate_proxy_methods.py
222
223 def __contains__(self, instance: object) -> bool:
224 r"""Return True if the instance is associated with this session.
225
226 .. container:: class_bases
227
228 Proxied for the :class:`_asyncio.AsyncSession` class on
229 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
230
231 .. container:: class_bases
232
233 Proxied for the :class:`_orm.Session` class on
234 behalf of the :class:`_asyncio.AsyncSession` class.
235
236 The instance may be pending or persistent within the Session for a
237 result of True.
238
239
240
241 """ # noqa: E501
242
243 return self._proxied.__contains__(instance)
244
245 def __iter__(self) -> Iterator[object]:
246 r"""Iterate over all pending or persistent instances within this
247 Session.
248
249 .. container:: class_bases
250
251 Proxied for the :class:`_asyncio.AsyncSession` class on
252 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
253
254 .. container:: class_bases
255
256 Proxied for the :class:`_orm.Session` class on
257 behalf of the :class:`_asyncio.AsyncSession` class.
258
259
260
261 """ # noqa: E501
262
263 return self._proxied.__iter__()
264
265 async def aclose(self) -> None:
266 r"""A synonym for :meth:`_asyncio.AsyncSession.close`.
267
268 .. container:: class_bases
269
270 Proxied for the :class:`_asyncio.AsyncSession` class on
271 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
272
273 The :meth:`_asyncio.AsyncSession.aclose` name is specifically
274 to support the Python standard library ``@contextlib.aclosing``
275 context manager function.
276
277 .. versionadded:: 2.0.20
278
279
280 """ # noqa: E501
281
282 return await self._proxied.aclose()
283
284 def add(self, instance: object, _warn: bool = True) -> None:
285 r"""Place an object into this :class:`_orm.Session`.
286
287 .. container:: class_bases
288
289 Proxied for the :class:`_asyncio.AsyncSession` class on
290 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
291
292 .. container:: class_bases
293
294 Proxied for the :class:`_orm.Session` class on
295 behalf of the :class:`_asyncio.AsyncSession` class.
296
297 Objects that are in the :term:`transient` state when passed to the
298 :meth:`_orm.Session.add` method will move to the
299 :term:`pending` state, until the next flush, at which point they
300 will move to the :term:`persistent` state.
301
302 Objects that are in the :term:`detached` state when passed to the
303 :meth:`_orm.Session.add` method will move to the :term:`persistent`
304 state directly.
305
306 If the transaction used by the :class:`_orm.Session` is rolled back,
307 objects which were transient when they were passed to
308 :meth:`_orm.Session.add` will be moved back to the
309 :term:`transient` state, and will no longer be present within this
310 :class:`_orm.Session`.
311
312 .. seealso::
313
314 :meth:`_orm.Session.add_all`
315
316 :ref:`session_adding` - at :ref:`session_basics`
317
318
319
320 """ # noqa: E501
321
322 return self._proxied.add(instance, _warn=_warn)
323
324 def add_all(self, instances: Iterable[object]) -> None:
325 r"""Add the given collection of instances to this :class:`_orm.Session`.
326
327 .. container:: class_bases
328
329 Proxied for the :class:`_asyncio.AsyncSession` class on
330 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
331
332 .. container:: class_bases
333
334 Proxied for the :class:`_orm.Session` class on
335 behalf of the :class:`_asyncio.AsyncSession` class.
336
337 See the documentation for :meth:`_orm.Session.add` for a general
338 behavioral description.
339
340 .. seealso::
341
342 :meth:`_orm.Session.add`
343
344 :ref:`session_adding` - at :ref:`session_basics`
345
346
347
348 """ # noqa: E501
349
350 return self._proxied.add_all(instances)
351
352 def begin(self) -> AsyncSessionTransaction:
353 r"""Return an :class:`_asyncio.AsyncSessionTransaction` object.
354
355 .. container:: class_bases
356
357 Proxied for the :class:`_asyncio.AsyncSession` class on
358 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
359
360 The underlying :class:`_orm.Session` will perform the
361 "begin" action when the :class:`_asyncio.AsyncSessionTransaction`
362 object is entered::
363
364 async with async_session.begin():
365 ... # ORM transaction is begun
366
367 Note that database IO will not normally occur when the session-level
368 transaction is begun, as database transactions begin on an
369 on-demand basis. However, the begin block is async to accommodate
370 for a :meth:`_orm.SessionEvents.after_transaction_create`
371 event hook that may perform IO.
372
373 For a general description of ORM begin, see
374 :meth:`_orm.Session.begin`.
375
376
377 """ # noqa: E501
378
379 return self._proxied.begin()
380
381 def begin_nested(self) -> AsyncSessionTransaction:
382 r"""Return an :class:`_asyncio.AsyncSessionTransaction` object
383 which will begin a "nested" transaction, e.g. SAVEPOINT.
384
385 .. container:: class_bases
386
387 Proxied for the :class:`_asyncio.AsyncSession` class on
388 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
389
390 Behavior is the same as that of :meth:`_asyncio.AsyncSession.begin`.
391
392 For a general description of ORM begin nested, see
393 :meth:`_orm.Session.begin_nested`.
394
395 .. seealso::
396
397 :ref:`aiosqlite_serializable` - special workarounds required
398 with the SQLite asyncio driver in order for SAVEPOINT to work
399 correctly.
400
401
402 """ # noqa: E501
403
404 return self._proxied.begin_nested()
405
406 async def close(self) -> None:
407 r"""Close out the transactional resources and ORM objects used by this
408 :class:`_asyncio.AsyncSession`.
409
410 .. container:: class_bases
411
412 Proxied for the :class:`_asyncio.AsyncSession` class on
413 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
414
415 .. seealso::
416
417 :meth:`_orm.Session.close` - main documentation for
418 "close"
419
420 :ref:`session_closing` - detail on the semantics of
421 :meth:`_asyncio.AsyncSession.close` and
422 :meth:`_asyncio.AsyncSession.reset`.
423
424
425 """ # noqa: E501
426
427 return await self._proxied.close()
428
429 async def reset(self) -> None:
430 r"""Close out the transactional resources and ORM objects used by this
431 :class:`_orm.Session`, resetting the session to its initial state.
432
433 .. container:: class_bases
434
435 Proxied for the :class:`_asyncio.AsyncSession` class on
436 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
437
438 .. versionadded:: 2.0.22
439
440 .. seealso::
441
442 :meth:`_orm.Session.reset` - main documentation for
443 "reset"
444
445 :ref:`session_closing` - detail on the semantics of
446 :meth:`_asyncio.AsyncSession.close` and
447 :meth:`_asyncio.AsyncSession.reset`.
448
449
450 """ # noqa: E501
451
452 return await self._proxied.reset()
453
454 async def commit(self) -> None:
455 r"""Commit the current transaction in progress.
456
457 .. container:: class_bases
458
459 Proxied for the :class:`_asyncio.AsyncSession` class on
460 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
461
462 .. seealso::
463
464 :meth:`_orm.Session.commit` - main documentation for
465 "commit"
466
467 """ # noqa: E501
468
469 return await self._proxied.commit()
470
471 async def connection(
472 self,
473 bind_arguments: Optional[_BindArguments] = None,
474 execution_options: Optional[CoreExecuteOptionsParameter] = None,
475 **kw: Any,
476 ) -> AsyncConnection:
477 r"""Return a :class:`_asyncio.AsyncConnection` object corresponding to
478 this :class:`.Session` object's transactional state.
479
480 .. container:: class_bases
481
482 Proxied for the :class:`_asyncio.AsyncSession` class on
483 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
484
485 This method may also be used to establish execution options for the
486 database connection used by the current transaction.
487
488 .. versionadded:: 1.4.24 Added \**kw arguments which are passed
489 through to the underlying :meth:`_orm.Session.connection` method.
490
491 .. seealso::
492
493 :meth:`_orm.Session.connection` - main documentation for
494 "connection"
495
496
497 """ # noqa: E501
498
499 return await self._proxied.connection(
500 bind_arguments=bind_arguments,
501 execution_options=execution_options,
502 **kw,
503 )
504
505 async def delete(self, instance: object) -> None:
506 r"""Mark an instance as deleted.
507
508 .. container:: class_bases
509
510 Proxied for the :class:`_asyncio.AsyncSession` class on
511 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
512
513 The database delete operation occurs upon ``flush()``.
514
515 As this operation may need to cascade along unloaded relationships,
516 it is awaitable to allow for those queries to take place.
517
518 .. seealso::
519
520 :meth:`_orm.Session.delete` - main documentation for delete
521
522
523 """ # noqa: E501
524
525 return await self._proxied.delete(instance)
526
527 @overload
528 async def execute(
529 self,
530 statement: TypedReturnsRows[_T],
531 params: Optional[_CoreAnyExecuteParams] = None,
532 *,
533 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
534 bind_arguments: Optional[_BindArguments] = None,
535 _parent_execute_state: Optional[Any] = None,
536 _add_event: Optional[Any] = None,
537 ) -> Result[_T]: ...
538
539 @overload
540 async def execute(
541 self,
542 statement: Executable,
543 params: Optional[_CoreAnyExecuteParams] = None,
544 *,
545 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
546 bind_arguments: Optional[_BindArguments] = None,
547 _parent_execute_state: Optional[Any] = None,
548 _add_event: Optional[Any] = None,
549 ) -> Result[Any]: ...
550
551 async def execute(
552 self,
553 statement: Executable,
554 params: Optional[_CoreAnyExecuteParams] = None,
555 *,
556 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
557 bind_arguments: Optional[_BindArguments] = None,
558 **kw: Any,
559 ) -> Result[Any]:
560 r"""Execute a statement and return a buffered
561 :class:`_engine.Result` object.
562
563 .. container:: class_bases
564
565 Proxied for the :class:`_asyncio.AsyncSession` class on
566 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
567
568 .. seealso::
569
570 :meth:`_orm.Session.execute` - main documentation for execute
571
572
573 """ # noqa: E501
574
575 return await self._proxied.execute(
576 statement,
577 params=params,
578 execution_options=execution_options,
579 bind_arguments=bind_arguments,
580 **kw,
581 )
582
583 def expire(
584 self, instance: object, attribute_names: Optional[Iterable[str]] = None
585 ) -> None:
586 r"""Expire the attributes on an instance.
587
588 .. container:: class_bases
589
590 Proxied for the :class:`_asyncio.AsyncSession` class on
591 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
592
593 .. container:: class_bases
594
595 Proxied for the :class:`_orm.Session` class on
596 behalf of the :class:`_asyncio.AsyncSession` class.
597
598 Marks the attributes of an instance as out of date. When an expired
599 attribute is next accessed, a query will be issued to the
600 :class:`.Session` object's current transactional context in order to
601 load all expired attributes for the given instance. Note that
602 a highly isolated transaction will return the same values as were
603 previously read in that same transaction, regardless of changes
604 in database state outside of that transaction.
605
606 To expire all objects in the :class:`.Session` simultaneously,
607 use :meth:`Session.expire_all`.
608
609 The :class:`.Session` object's default behavior is to
610 expire all state whenever the :meth:`Session.rollback`
611 or :meth:`Session.commit` methods are called, so that new
612 state can be loaded for the new transaction. For this reason,
613 calling :meth:`Session.expire` only makes sense for the specific
614 case that a non-ORM SQL statement was emitted in the current
615 transaction.
616
617 :param instance: The instance to be refreshed.
618 :param attribute_names: optional list of string attribute names
619 indicating a subset of attributes to be expired.
620
621 .. seealso::
622
623 :ref:`session_expire` - introductory material
624
625 :meth:`.Session.expire`
626
627 :meth:`.Session.refresh`
628
629 :meth:`_orm.Query.populate_existing`
630
631
632
633 """ # noqa: E501
634
635 return self._proxied.expire(instance, attribute_names=attribute_names)
636
637 def expire_all(self) -> None:
638 r"""Expires all persistent instances within this Session.
639
640 .. container:: class_bases
641
642 Proxied for the :class:`_asyncio.AsyncSession` class on
643 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
644
645 .. container:: class_bases
646
647 Proxied for the :class:`_orm.Session` class on
648 behalf of the :class:`_asyncio.AsyncSession` class.
649
650 When any attributes on a persistent instance is next accessed,
651 a query will be issued using the
652 :class:`.Session` object's current transactional context in order to
653 load all expired attributes for the given instance. Note that
654 a highly isolated transaction will return the same values as were
655 previously read in that same transaction, regardless of changes
656 in database state outside of that transaction.
657
658 To expire individual objects and individual attributes
659 on those objects, use :meth:`Session.expire`.
660
661 The :class:`.Session` object's default behavior is to
662 expire all state whenever the :meth:`Session.rollback`
663 or :meth:`Session.commit` methods are called, so that new
664 state can be loaded for the new transaction. For this reason,
665 calling :meth:`Session.expire_all` is not usually needed,
666 assuming the transaction is isolated.
667
668 .. seealso::
669
670 :ref:`session_expire` - introductory material
671
672 :meth:`.Session.expire`
673
674 :meth:`.Session.refresh`
675
676 :meth:`_orm.Query.populate_existing`
677
678
679
680 """ # noqa: E501
681
682 return self._proxied.expire_all()
683
684 def expunge(self, instance: object) -> None:
685 r"""Remove the `instance` from this ``Session``.
686
687 .. container:: class_bases
688
689 Proxied for the :class:`_asyncio.AsyncSession` class on
690 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
691
692 .. container:: class_bases
693
694 Proxied for the :class:`_orm.Session` class on
695 behalf of the :class:`_asyncio.AsyncSession` class.
696
697 This will free all internal references to the instance. Cascading
698 will be applied according to the *expunge* cascade rule.
699
700
701
702 """ # noqa: E501
703
704 return self._proxied.expunge(instance)
705
706 def expunge_all(self) -> None:
707 r"""Remove all object instances from this ``Session``.
708
709 .. container:: class_bases
710
711 Proxied for the :class:`_asyncio.AsyncSession` class on
712 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
713
714 .. container:: class_bases
715
716 Proxied for the :class:`_orm.Session` class on
717 behalf of the :class:`_asyncio.AsyncSession` class.
718
719 This is equivalent to calling ``expunge(obj)`` on all objects in this
720 ``Session``.
721
722
723
724 """ # noqa: E501
725
726 return self._proxied.expunge_all()
727
728 async def flush(self, objects: Optional[Sequence[Any]] = None) -> None:
729 r"""Flush all the object changes to the database.
730
731 .. container:: class_bases
732
733 Proxied for the :class:`_asyncio.AsyncSession` class on
734 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
735
736 .. seealso::
737
738 :meth:`_orm.Session.flush` - main documentation for flush
739
740
741 """ # noqa: E501
742
743 return await self._proxied.flush(objects=objects)
744
745 def get_bind(
746 self,
747 mapper: Optional[_EntityBindKey[_O]] = None,
748 clause: Optional[ClauseElement] = None,
749 bind: Optional[_SessionBind] = None,
750 **kw: Any,
751 ) -> Union[Engine, Connection]:
752 r"""Return a "bind" to which the synchronous proxied :class:`_orm.Session`
753 is bound.
754
755 .. container:: class_bases
756
757 Proxied for the :class:`_asyncio.AsyncSession` class on
758 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
759
760 Unlike the :meth:`_orm.Session.get_bind` method, this method is
761 currently **not** used by this :class:`.AsyncSession` in any way
762 in order to resolve engines for requests.
763
764 .. note::
765
766 This method proxies directly to the :meth:`_orm.Session.get_bind`
767 method, however is currently **not** useful as an override target,
768 in contrast to that of the :meth:`_orm.Session.get_bind` method.
769 The example below illustrates how to implement custom
770 :meth:`_orm.Session.get_bind` schemes that work with
771 :class:`.AsyncSession` and :class:`.AsyncEngine`.
772
773 The pattern introduced at :ref:`session_custom_partitioning`
774 illustrates how to apply a custom bind-lookup scheme to a
775 :class:`_orm.Session` given a set of :class:`_engine.Engine` objects.
776 To apply a corresponding :meth:`_orm.Session.get_bind` implementation
777 for use with a :class:`.AsyncSession` and :class:`.AsyncEngine`
778 objects, continue to subclass :class:`_orm.Session` and apply it to
779 :class:`.AsyncSession` using
780 :paramref:`.AsyncSession.sync_session_class`. The inner method must
781 continue to return :class:`_engine.Engine` instances, which can be
782 acquired from a :class:`_asyncio.AsyncEngine` using the
783 :attr:`_asyncio.AsyncEngine.sync_engine` attribute::
784
785 # using example from "Custom Vertical Partitioning"
786
787
788 import random
789
790 from sqlalchemy.ext.asyncio import AsyncSession
791 from sqlalchemy.ext.asyncio import create_async_engine
792 from sqlalchemy.ext.asyncio import async_sessionmaker
793 from sqlalchemy.orm import Session
794
795 # construct async engines w/ async drivers
796 engines = {
797 "leader": create_async_engine("sqlite+aiosqlite:///leader.db"),
798 "other": create_async_engine("sqlite+aiosqlite:///other.db"),
799 "follower1": create_async_engine("sqlite+aiosqlite:///follower1.db"),
800 "follower2": create_async_engine("sqlite+aiosqlite:///follower2.db"),
801 }
802
803
804 class RoutingSession(Session):
805 def get_bind(self, mapper=None, clause=None, **kw):
806 # within get_bind(), return sync engines
807 if mapper and issubclass(mapper.class_, MyOtherClass):
808 return engines["other"].sync_engine
809 elif self._flushing or isinstance(clause, (Update, Delete)):
810 return engines["leader"].sync_engine
811 else:
812 return engines[
813 random.choice(["follower1", "follower2"])
814 ].sync_engine
815
816
817 # apply to AsyncSession using sync_session_class
818 AsyncSessionMaker = async_sessionmaker(sync_session_class=RoutingSession)
819
820 The :meth:`_orm.Session.get_bind` method is called in a non-asyncio,
821 implicitly non-blocking context in the same manner as ORM event hooks
822 and functions that are invoked via :meth:`.AsyncSession.run_sync`, so
823 routines that wish to run SQL commands inside of
824 :meth:`_orm.Session.get_bind` can continue to do so using
825 blocking-style code, which will be translated to implicitly async calls
826 at the point of invoking IO on the database drivers.
827
828
829 """ # noqa: E501
830
831 return self._proxied.get_bind(
832 mapper=mapper, clause=clause, bind=bind, **kw
833 )
834
835 def is_modified(
836 self, instance: object, include_collections: bool = True
837 ) -> bool:
838 r"""Return ``True`` if the given instance has locally
839 modified attributes.
840
841 .. container:: class_bases
842
843 Proxied for the :class:`_asyncio.AsyncSession` class on
844 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
845
846 .. container:: class_bases
847
848 Proxied for the :class:`_orm.Session` class on
849 behalf of the :class:`_asyncio.AsyncSession` class.
850
851 This method retrieves the history for each instrumented
852 attribute on the instance and performs a comparison of the current
853 value to its previously flushed or committed value, if any.
854
855 It is in effect a more expensive and accurate
856 version of checking for the given instance in the
857 :attr:`.Session.dirty` collection; a full test for
858 each attribute's net "dirty" status is performed.
859
860 E.g.::
861
862 return session.is_modified(someobject)
863
864 A few caveats to this method apply:
865
866 * Instances present in the :attr:`.Session.dirty` collection may
867 report ``False`` when tested with this method. This is because
868 the object may have received change events via attribute mutation,
869 thus placing it in :attr:`.Session.dirty`, but ultimately the state
870 is the same as that loaded from the database, resulting in no net
871 change here.
872 * Scalar attributes may not have recorded the previously set
873 value when a new value was applied, if the attribute was not loaded,
874 or was expired, at the time the new value was received - in these
875 cases, the attribute is assumed to have a change, even if there is
876 ultimately no net change against its database value. SQLAlchemy in
877 most cases does not need the "old" value when a set event occurs, so
878 it skips the expense of a SQL call if the old value isn't present,
879 based on the assumption that an UPDATE of the scalar value is
880 usually needed, and in those few cases where it isn't, is less
881 expensive on average than issuing a defensive SELECT.
882
883 The "old" value is fetched unconditionally upon set only if the
884 attribute container has the ``active_history`` flag set to ``True``.
885 This flag is set typically for primary key attributes and scalar
886 object references that are not a simple many-to-one. To set this
887 flag for any arbitrary mapped column, use the ``active_history``
888 argument with :func:`.column_property`.
889
890 :param instance: mapped instance to be tested for pending changes.
891 :param include_collections: Indicates if multivalued collections
892 should be included in the operation. Setting this to ``False`` is a
893 way to detect only local-column based properties (i.e. scalar columns
894 or many-to-one foreign keys) that would result in an UPDATE for this
895 instance upon flush.
896
897
898
899 """ # noqa: E501
900
901 return self._proxied.is_modified(
902 instance, include_collections=include_collections
903 )
904
905 async def invalidate(self) -> None:
906 r"""Close this Session, using connection invalidation.
907
908 .. container:: class_bases
909
910 Proxied for the :class:`_asyncio.AsyncSession` class on
911 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
912
913 For a complete description, see :meth:`_orm.Session.invalidate`.
914
915 """ # noqa: E501
916
917 return await self._proxied.invalidate()
918
919 async def merge(
920 self,
921 instance: _O,
922 *,
923 load: bool = True,
924 options: Optional[Sequence[ORMOption]] = None,
925 ) -> _O:
926 r"""Copy the state of a given instance into a corresponding instance
927 within this :class:`_asyncio.AsyncSession`.
928
929 .. container:: class_bases
930
931 Proxied for the :class:`_asyncio.AsyncSession` class on
932 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
933
934 .. seealso::
935
936 :meth:`_orm.Session.merge` - main documentation for merge
937
938
939 """ # noqa: E501
940
941 return await self._proxied.merge(instance, load=load, options=options)
942
943 async def refresh(
944 self,
945 instance: object,
946 attribute_names: Optional[Iterable[str]] = None,
947 with_for_update: ForUpdateParameter = None,
948 ) -> None:
949 r"""Expire and refresh the attributes on the given instance.
950
951 .. container:: class_bases
952
953 Proxied for the :class:`_asyncio.AsyncSession` class on
954 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
955
956 A query will be issued to the database and all attributes will be
957 refreshed with their current database value.
958
959 This is the async version of the :meth:`_orm.Session.refresh` method.
960 See that method for a complete description of all options.
961
962 .. seealso::
963
964 :meth:`_orm.Session.refresh` - main documentation for refresh
965
966
967 """ # noqa: E501
968
969 return await self._proxied.refresh(
970 instance,
971 attribute_names=attribute_names,
972 with_for_update=with_for_update,
973 )
974
975 async def rollback(self) -> None:
976 r"""Rollback the current transaction in progress.
977
978 .. container:: class_bases
979
980 Proxied for the :class:`_asyncio.AsyncSession` class on
981 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
982
983 .. seealso::
984
985 :meth:`_orm.Session.rollback` - main documentation for
986 "rollback"
987
988 """ # noqa: E501
989
990 return await self._proxied.rollback()
991
992 @overload
993 async def scalar(
994 self,
995 statement: TypedReturnsRows[Tuple[_T]],
996 params: Optional[_CoreAnyExecuteParams] = None,
997 *,
998 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
999 bind_arguments: Optional[_BindArguments] = None,
1000 **kw: Any,
1001 ) -> Optional[_T]: ...
1002
1003 @overload
1004 async def scalar(
1005 self,
1006 statement: Executable,
1007 params: Optional[_CoreAnyExecuteParams] = None,
1008 *,
1009 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1010 bind_arguments: Optional[_BindArguments] = None,
1011 **kw: Any,
1012 ) -> Any: ...
1013
1014 async def scalar(
1015 self,
1016 statement: Executable,
1017 params: Optional[_CoreAnyExecuteParams] = None,
1018 *,
1019 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1020 bind_arguments: Optional[_BindArguments] = None,
1021 **kw: Any,
1022 ) -> Any:
1023 r"""Execute a statement and return a scalar result.
1024
1025 .. container:: class_bases
1026
1027 Proxied for the :class:`_asyncio.AsyncSession` class on
1028 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
1029
1030 .. seealso::
1031
1032 :meth:`_orm.Session.scalar` - main documentation for scalar
1033
1034
1035 """ # noqa: E501
1036
1037 return await self._proxied.scalar(
1038 statement,
1039 params=params,
1040 execution_options=execution_options,
1041 bind_arguments=bind_arguments,
1042 **kw,
1043 )
1044
1045 @overload
1046 async def scalars(
1047 self,
1048 statement: TypedReturnsRows[Tuple[_T]],
1049 params: Optional[_CoreAnyExecuteParams] = None,
1050 *,
1051 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1052 bind_arguments: Optional[_BindArguments] = None,
1053 **kw: Any,
1054 ) -> ScalarResult[_T]: ...
1055
1056 @overload
1057 async def scalars(
1058 self,
1059 statement: Executable,
1060 params: Optional[_CoreAnyExecuteParams] = None,
1061 *,
1062 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1063 bind_arguments: Optional[_BindArguments] = None,
1064 **kw: Any,
1065 ) -> ScalarResult[Any]: ...
1066
1067 async def scalars(
1068 self,
1069 statement: Executable,
1070 params: Optional[_CoreAnyExecuteParams] = None,
1071 *,
1072 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1073 bind_arguments: Optional[_BindArguments] = None,
1074 **kw: Any,
1075 ) -> ScalarResult[Any]:
1076 r"""Execute a statement and return scalar results.
1077
1078 .. container:: class_bases
1079
1080 Proxied for the :class:`_asyncio.AsyncSession` class on
1081 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
1082
1083 :return: a :class:`_result.ScalarResult` object
1084
1085 .. versionadded:: 1.4.24 Added :meth:`_asyncio.AsyncSession.scalars`
1086
1087 .. versionadded:: 1.4.26 Added
1088 :meth:`_asyncio.async_scoped_session.scalars`
1089
1090 .. seealso::
1091
1092 :meth:`_orm.Session.scalars` - main documentation for scalars
1093
1094 :meth:`_asyncio.AsyncSession.stream_scalars` - streaming version
1095
1096
1097 """ # noqa: E501
1098
1099 return await self._proxied.scalars(
1100 statement,
1101 params=params,
1102 execution_options=execution_options,
1103 bind_arguments=bind_arguments,
1104 **kw,
1105 )
1106
1107 async def get(
1108 self,
1109 entity: _EntityBindKey[_O],
1110 ident: _PKIdentityArgument,
1111 *,
1112 options: Optional[Sequence[ORMOption]] = None,
1113 populate_existing: bool = False,
1114 with_for_update: ForUpdateParameter = None,
1115 identity_token: Optional[Any] = None,
1116 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1117 ) -> Union[_O, None]:
1118 r"""Return an instance based on the given primary key identifier,
1119 or ``None`` if not found.
1120
1121 .. container:: class_bases
1122
1123 Proxied for the :class:`_asyncio.AsyncSession` class on
1124 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
1125
1126 .. seealso::
1127
1128 :meth:`_orm.Session.get` - main documentation for get
1129
1130
1131
1132 """ # noqa: E501
1133
1134 result = await self._proxied.get(
1135 entity,
1136 ident,
1137 options=options,
1138 populate_existing=populate_existing,
1139 with_for_update=with_for_update,
1140 identity_token=identity_token,
1141 execution_options=execution_options,
1142 )
1143 return result
1144
1145 async def get_one(
1146 self,
1147 entity: _EntityBindKey[_O],
1148 ident: _PKIdentityArgument,
1149 *,
1150 options: Optional[Sequence[ORMOption]] = None,
1151 populate_existing: bool = False,
1152 with_for_update: ForUpdateParameter = None,
1153 identity_token: Optional[Any] = None,
1154 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1155 ) -> _O:
1156 r"""Return an instance based on the given primary key identifier,
1157 or raise an exception if not found.
1158
1159 .. container:: class_bases
1160
1161 Proxied for the :class:`_asyncio.AsyncSession` class on
1162 behalf of the :class:`_asyncio.scoping.async_scoped_session` class.
1163
1164 Raises :class:`_exc.NoResultFound` if the query selects no rows.
1165
1166 ..versionadded: 2.0.22
1167
1168 .. seealso::
1169
1170 :meth:`_orm.Session.get_one` - main documentation for get_one
1171
1172
1173 """ # noqa: E501
1174
1175 return await self._proxied.get_one(
1176 entity,
1177 ident,
1178 options=options,
1179 populate_existing=populate_existing,
1180 with_for_update=with_for_update,
1181 identity_token=identity_token,
1182 execution_options=execution_options,
1183 )
1184
1185 @overload
1186 async def stream(
1187 self,
1188 statement: TypedReturnsRows[_T],
1189 params: Optional[_CoreAnyExecuteParams] = None,
1190 *,
1191 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
1192 bind_arguments: Optional[_BindArguments] = None,
1193 **kw: Any,
1194 ) -> AsyncResult[_T]: ...
1195
1196 @overload
1197 async def stream(
1198 self,
1199 statement: Executable,
1200 params: Optional[_CoreAnyExecuteParams] = None,