perf: Add the scoped attribute to determine whether the session attribute is scoped.

Author: amisadmin

sqlalchemy_database/database.py

@@ -107,6 +107,13 @@ async def get_user(id:int):
         """
         return self.scoped_session()
 
+    @property
+    def scoped(self) -> bool:
+        """Whether the current context has a session. If False, the session is the default global session,
+        and the transaction needs to be manually submitted.
+        """
+        return bool(self._session_scope.get())
+
     def __call__(self, scope: Any = None):
         return AsyncSessionContextVarManager(self, scope=scope)
 
@@ -140,7 +147,7 @@ async def session_generator(self) -> AsyncGenerator[AsyncSession, Any]:
                 return await session.get(User,id)
             ```
         """
-        if self._session_scope.get():
+        if self.scoped:
             """If the current context has a session, return it."""
             yield self.session
         else:
@@ -210,6 +217,10 @@ def __init__(self, engine: Engine, commit_on_exit: bool = True, **session_option
     def session(self) -> Session:
         return self.scoped_session()
 
+    @property
+    def scoped(self) -> bool:
+        return bool(self._session_scope.get())
+
     def __call__(self, scope: Any = None):
         return SessionContextVarManager(self, scope=scope)
 
@@ -223,7 +234,7 @@ def create(
         return cls(engine, **session_options)
 
     def session_generator(self) -> Generator[Session, Any, None]:
-        if self._session_scope.get():
+        if self.scoped:
             """If the current context has a session, return it."""
             yield self.session
         else:

tests/test_AsyncDatabase.py

@@ -98,8 +98,9 @@ def lock(event_loop: AbstractEventLoop):
 
 async def test_async_session_context_var(lock, i=1):
     global_session = async_db.session  # Default global session
-
+    assert not async_db.scoped
     async with async_db() as session:  # Enter a new session
+        assert async_db.scoped
         user = await session.get(User, 1)
         assert user.id == 1
         assert session is async_db.session

tests/test_Database.py

@@ -84,7 +84,9 @@ def test_sqlmodel_session():
 
 def test_session_context_var(i=1):
     global_session = sync_db.session  # Default global session
+    assert not sync_db.scoped
     with sync_db() as session:  # Enter a new session
+        assert sync_db.scoped
         user = session.get(User, 1)
         assert user.id == 1
         assert session is sync_db.session