diff --git a/README.md b/README.md index 43dbe69..80da903 100644 --- a/README.md +++ b/README.md @@ -89,81 +89,93 @@ adapter = casbin_async_sqlalchemy_adapter.Adapter( e = casbin.AsyncEnforcer('path/to/model.conf', adapter) ``` -## External Session Support +## Atomic Transactions (Transaction Control) -The adapter supports using externally managed SQLAlchemy sessions. This feature is useful for: +The adapter lets you group multiple enforcer or adapter calls into a single atomic database transaction, so that either **all changes are committed together or none are**. -- Better transaction control in complex scenarios -- Reducing database connections and communications -- Supporting advanced database features like two-phase commits -- Integrating with existing database session management +### Using `adapter.transaction()` — recommended for enforcer-level calls -### Basic Usage with External Session +The `transaction()` context manager binds a session to every adapter operation that happens inside the block. You can obtain the adapter from an existing enforcer with `enforcer.get_adapter()`, so there is no need for a separate import or variable in most cases. ```python -import casbin_async_sqlalchemy_adapter import casbin -from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession -from sqlalchemy.orm import sessionmaker +import casbin_async_sqlalchemy_adapter +from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker -# Create your own database session engine = create_async_engine('sqlite+aiosqlite:///test.db') -async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) +async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) -# Create adapter with external session -session = async_session() -adapter = casbin_async_sqlalchemy_adapter.Adapter( - 'sqlite+aiosqlite:///test.db', - db_session=session -) +adapter = casbin_async_sqlalchemy_adapter.Adapter(engine) +await adapter.create_table() -e = casbin.AsyncEnforcer('path/to/model.conf', adapter) +enforcer = casbin.AsyncEnforcer('path/to/model.conf', adapter) +await enforcer.load_policy() -# Now you have full control over the session -# The adapter will not auto-commit or auto-rollback when using external sessions +# --- elsewhere in the application --- + +# Obtain the adapter directly from the enforcer — no separate import needed. +adapter = enforcer.get_adapter() + +async with async_session() as session: + async with adapter.transaction(session=session): + # All enforcer calls inside this block share `session` and do NOT + # auto-commit — the adapter defers commit control to the caller. + await enforcer.add_policy("alice", "data1", "read") + await enforcer.add_policy("bob", "data2", "write") + + # Commit or roll back the entire batch atomically. + await session.commit() # persists both policies + # await session.rollback() # would discard both +``` + +#### Atomic user-creation example + +```python +async def create_user(session, user_data): + # Insert the user row – not yet committed. + user = User(**user_data) + session.add(user) + await session.flush() # get user.id without committing + + adapter = enforcer.get_adapter() + async with adapter.transaction(session=session): + await enforcer.add_role_for_user(str(user.id), "viewer") + await enforcer.add_policy(str(user.id), "resource", "read") + + # One commit covers the user row AND all policy changes. + await session.commit() + # If anything raises before this point, session.rollback() leaves + # the database in a clean state — no orphaned policies. ``` -### Transaction Control Example +### Per-method `session` and `commit` parameters + +For direct adapter calls you can pass `session` and `commit` keyword arguments individually: ```python -# Example: Manual transaction control async with async_session() as session: - adapter = casbin_async_sqlalchemy_adapter.Adapter( - 'sqlite+aiosqlite:///test.db', - db_session=session - ) - - e = casbin.AsyncEnforcer('path/to/model.conf', adapter) - - # Add multiple policies in a single transaction - await e.add_policy("alice", "data1", "read") - await e.add_policy("bob", "data2", "write") - - # Commit or rollback as needed + await adapter.add_policy("p", "p", ["alice", "data1", "read"], + session=session, commit=False) + await adapter.add_policies("p", "p", [["bob", "data2", "write"]], + session=session, commit=False) await session.commit() ``` -### Batch Operations Example +All write methods accept these parameters: `add_policy`, `add_policies`, `remove_policy`, `remove_policies`, `remove_filtered_policy`, `update_policy`, `update_policies`, `update_filtered_policies`. + +### Constructor-level external session (`db_session`) + +The original constructor-level session is still supported for scenarios where every operation should share the same long-lived session: ```python -# Example: Efficient batch operations async with async_session() as session: - adapter = casbin_async_sqlalchemy_adapter.Adapter( - 'sqlite+aiosqlite:///test.db', - db_session=session - ) - + adapter = casbin_async_sqlalchemy_adapter.Adapter(engine, db_session=session) e = casbin.AsyncEnforcer('path/to/model.conf', adapter) - - # Batch add multiple policies efficiently - policies = [ - ["alice", "data1", "read"], - ["bob", "data2", "write"], - ["carol", "data3", "read"] - ] - await e.add_policies(policies) - - # Commit the transaction + await e.load_policy() + + await e.add_policy("alice", "data1", "read") + await e.add_policy("bob", "data2", "write") + await session.commit() ``` diff --git a/casbin_async_sqlalchemy_adapter/adapter.py b/casbin_async_sqlalchemy_adapter/adapter.py index 884103f..853dfac 100644 --- a/casbin_async_sqlalchemy_adapter/adapter.py +++ b/casbin_async_sqlalchemy_adapter/adapter.py @@ -12,8 +12,12 @@ # See the License for the specific language governing permissions and # limitations under the License. from contextlib import asynccontextmanager +from contextvars import ContextVar from typing import List, Optional +# Sentinel used to distinguish "not inside a transaction() context" from an explicit session=None. +_UNSET = object() + from casbin import persist from casbin.persist.adapters.asyncio import AsyncAdapter from sqlalchemy import Column, Integer, String, Boolean, delete, insert @@ -137,21 +141,76 @@ def __init__( self._filtered = filtered + # Per-task ContextVars for the transaction() context manager. + # These are instance-level to avoid cross-adapter interference. + self._tx_session_var: ContextVar = ContextVar(f"_tx_session_{id(self)}", default=_UNSET) + self._tx_commit_var: ContextVar[bool] = ContextVar(f"_tx_commit_{id(self)}", default=True) + @asynccontextmanager - async def _session_scope(self): + async def transaction(self, session: Optional[AsyncSession] = None, commit: bool = True): + """Bind a session to all adapter/enforcer operations within this block. + + This enables using high-level enforcer methods while controlling the + surrounding transaction externally, without accessing the adapter directly + for each individual call:: + + async with adapter.transaction(session=my_session): + await enforcer.add_policy(...) + await enforcer.add_policies(...) + await my_session.commit() # or await my_session.rollback() + + When *session* is ``None`` the ``commit`` parameter controls whether + internally-created sessions are auto-committed (mirrors the per-method + ``commit`` flag but applied to every call inside the block). + """ + token_s = self._tx_session_var.set(session) + token_c = self._tx_commit_var.set(commit) + try: + yield + finally: + self._tx_session_var.reset(token_s) + self._tx_commit_var.reset(token_c) + + @asynccontextmanager + async def _session_scope(self, commit: bool = True, session: Optional[AsyncSession] = None): """Provide an asynchronous transactional scope around a series of operations.""" + if session is not None: + # Explicit session passed by the caller (e.g. internal method chaining): use as-is. + yield session + return + + tx_val = self._tx_session_var.get() + if tx_val is not _UNSET: + # Inside a transaction() context manager. + if tx_val is not None: + # A real session was provided to transaction(): yield it without auto-commit. + yield tx_val + else: + # transaction(session=None): create an internal session governed by tx_commit. + async with self.session_local() as s: + try: + yield s + if self._tx_commit_var.get(): + await s.commit() + except Exception as e: + await s.rollback() + raise e + return + if self._external_session is not None: - # Use external session without automatic commit/rollback + # Constructor-level external session: yield without auto-commit. yield self._external_session - else: - # Use internal session with automatic commit/rollback - async with self.session_local() as session: - try: - yield session - await session.commit() - except Exception as e: - await session.rollback() - raise e + return + + # Default: create a short-lived internal session with auto-commit controlled by `commit`. + async with self.session_local() as s: + try: + yield s + if commit: + await s.commit() + except Exception as e: + await s.rollback() + raise e async def create_table(self): """Creates default casbin rule table.""" @@ -193,20 +252,12 @@ def _softdelete_query(self, stmt): stmt = stmt.where(not_(self.softdelete_attribute)) return stmt - async def _save_policy_line(self, ptype, rule, session=None): - if session is not None: - # Use provided session - line = self._db_class(ptype=ptype) - for i, v in enumerate(rule): - setattr(line, "v{}".format(i), v) - session.add(line) - else: - # Use session scope (for backward compatibility) - async with self._session_scope() as session: - line = self._db_class(ptype=ptype) - for i, v in enumerate(rule): - setattr(line, "v{}".format(i), v) - session.add(line) + def _save_policy_line(self, ptype, rule, session): + """Add a single policy line to the database within the given session.""" + line = self._db_class(ptype=ptype) + for i, v in enumerate(rule): + setattr(line, "v{}".format(i), v) + session.add(line) async def save_policy(self, model): """saves all policy rules to the storage.""" @@ -220,7 +271,7 @@ async def save_policy(self, model): continue for ptype, ast in model.model[sec].items(): for rule in ast.policy: - await self._save_policy_line(ptype, rule, session) + self._save_policy_line(ptype, rule, session) return True # Custom strategy for softdelete since it does not make sense to recreate all of the @@ -248,7 +299,7 @@ async def save_policy(self, model): # If the rule is not present, create an entry in the database result = await session.execute(filter_stmt) if result.scalar_one_or_none() is None: - await self._save_policy_line(ptype, rule, session=session) + self._save_policy_line(ptype, rule, session=session) for line in lines_before_changes: ptype = line.ptype @@ -292,11 +343,12 @@ async def clear_policy(self): setattr(line, self.softdelete_attribute.name, True) return True - async def add_policy(self, sec, ptype, rule): + async def add_policy(self, sec, ptype, rule, session: Optional[AsyncSession] = None, commit: bool = True): """adds a policy rule to the storage.""" - await self._save_policy_line(ptype, rule) + async with self._session_scope(commit=commit, session=session) as s: + self._save_policy_line(ptype, rule, s) - async def add_policies(self, sec, ptype, rules): + async def add_policies(self, sec, ptype, rules, session: Optional[AsyncSession] = None, commit: bool = True): """adds a policy rules to the storage.""" if not rules: return @@ -309,57 +361,57 @@ async def add_policies(self, sec, ptype, rules): row[f"v{i}"] = v rows.append(row) - async with self._session_scope() as session: + async with self._session_scope(commit=commit, session=session) as s: stmt = insert(self._db_class) - await session.execute(stmt, rows) + await s.execute(stmt, rows) - async def remove_policy(self, sec, ptype, rule): + async def remove_policy(self, sec, ptype, rule, session: Optional[AsyncSession] = None, commit: bool = True): """removes a policy rule from the storage.""" - async with self._session_scope() as session: + async with self._session_scope(commit=commit, session=session) as s: if self.softdelete_attribute is None: stmt = delete(self._db_class).where(self._db_class.ptype == ptype) for i, v in enumerate(rule): stmt = stmt.where(getattr(self._db_class, "v{}".format(i)) == v) - r = await session.execute(stmt) + r = await s.execute(stmt) return True if r.rowcount > 0 else False else: stmt = select(self._db_class).where(self._db_class.ptype == ptype) stmt = self._softdelete_query(stmt) for i, v in enumerate(rule): stmt = stmt.where(getattr(self._db_class, "v{}".format(i)) == v) - result = await session.execute(stmt) + result = await s.execute(stmt) lines = result.scalars().all() for line in lines: setattr(line, self.softdelete_attribute.name, True) return True if len(lines) > 0 else False - async def remove_policies(self, sec, ptype, rules): + async def remove_policies(self, sec, ptype, rules, session: Optional[AsyncSession] = None, commit: bool = True): """remove policy rules from the storage.""" if not rules: return - async with self._session_scope() as session: + async with self._session_scope(commit=commit, session=session) as s: if self.softdelete_attribute is None: stmt = delete(self._db_class).where(self._db_class.ptype == ptype) rules_zipped = zip(*rules) for i, rule in enumerate(rules_zipped): stmt = stmt.where(or_(getattr(self._db_class, "v{}".format(i)) == v for v in rule)) - await session.execute(stmt) + await s.execute(stmt) else: stmt = select(self._db_class).where(self._db_class.ptype == ptype) stmt = self._softdelete_query(stmt) rules_zipped = zip(*rules) for i, rule in enumerate(rules_zipped): stmt = stmt.where(or_(getattr(self._db_class, "v{}".format(i)) == v for v in rule)) - result = await session.execute(stmt) + result = await s.execute(stmt) lines = result.scalars().all() for line in lines: setattr(line, self.softdelete_attribute.name, True) - async def remove_filtered_policy(self, sec, ptype, field_index, *field_values): + async def remove_filtered_policy(self, sec, ptype, field_index, *field_values, session: Optional[AsyncSession] = None, commit: bool = True): """removes policy rules that match the filter from the storage. This is part of the Auto-Save feature. """ - async with self._session_scope() as session: + async with self._session_scope(commit=commit, session=session) as s: if not (0 <= field_index <= 5): return False if not (1 <= field_index + len(field_values) <= 6): @@ -371,7 +423,7 @@ async def remove_filtered_policy(self, sec, ptype, field_index, *field_values): if v != "": v_value = getattr(self._db_class, "v{}".format(field_index + i)) stmt = stmt.where(v_value == v) - r = await session.execute(stmt) + r = await s.execute(stmt) return True if r.rowcount > 0 else False else: stmt = select(self._db_class).where(self._db_class.ptype == ptype) @@ -380,13 +432,15 @@ async def remove_filtered_policy(self, sec, ptype, field_index, *field_values): if v != "": v_value = getattr(self._db_class, "v{}".format(field_index + i)) stmt = stmt.where(v_value == v) - result = await session.execute(stmt) + result = await s.execute(stmt) lines = result.scalars().all() for line in lines: setattr(line, self.softdelete_attribute.name, True) return True if len(lines) > 0 else False - async def update_policy(self, sec: str, ptype: str, old_rule: List[str], new_rule: List[str]) -> None: + async def update_policy( + self, sec: str, ptype: str, old_rule: List[str], new_rule: List[str], session: Optional[AsyncSession] = None, commit: bool = True + ) -> None: """ Update the old_rule with the new_rule in the database (storage). @@ -394,11 +448,13 @@ async def update_policy(self, sec: str, ptype: str, old_rule: List[str], new_rul :param ptype: policy type :param old_rule: the old rule that needs to be modified :param new_rule: the new rule to replace the old rule + :param session: optional external session to use + :param commit: whether to commit the transaction (only used when no session is provided) :return: None """ - async with self._session_scope() as session: + async with self._session_scope(commit=commit, session=session) as s: stmt = select(self._db_class).where(self._db_class.ptype == ptype) stmt = self._softdelete_query(stmt) @@ -409,7 +465,7 @@ async def update_policy(self, sec: str, ptype: str, old_rule: List[str], new_rul # need the length of the longest_rule to perform overwrite longest_rule = old_rule if len(old_rule) > len(new_rule) else new_rule - result = await session.execute(stmt) + result = await s.execute(stmt) old_rule_line = result.scalar_one() # overwrite the old rule with the new rule @@ -425,6 +481,8 @@ async def update_policies( ptype: str, old_rules: List[List[str]], new_rules: List[List[str]], + session: Optional[AsyncSession] = None, + commit: bool = True, ) -> None: """ Update the old_rules with the new_rules in the database (storage). @@ -433,13 +491,18 @@ async def update_policies( :param ptype: policy type :param old_rules: the old rules that need to be modified :param new_rules: the new rules to replace the old rules + :param session: optional external session to use + :param commit: whether to commit the transaction (only used when no session is provided) :return: None """ - for i in range(len(old_rules)): - await self.update_policy(sec, ptype, old_rules[i], new_rules[i]) + async with self._session_scope(commit=commit, session=session) as s: + for i in range(len(old_rules)): + await self.update_policy(sec, ptype, old_rules[i], new_rules[i], session=s, commit=False) - async def update_filtered_policies(self, sec, ptype, new_rules: List[List[str]], field_index, *field_values) -> List[List[str]]: + async def update_filtered_policies( + self, sec, ptype, new_rules: List[List[str]], field_index, *field_values, session: Optional[AsyncSession] = None, commit: bool = True + ) -> List[List[str]]: """update_filtered_policies updates all the policies on the basis of the filter.""" filter = Filter() @@ -452,18 +515,18 @@ async def update_filtered_policies(self, sec, ptype, new_rules: List[List[str]], else: break - return await self._update_filtered_policies(new_rules, filter) + return await self._update_filtered_policies(new_rules, filter, session=session, commit=commit) - async def _update_filtered_policies(self, new_rules, filter) -> List[List[str]]: + async def _update_filtered_policies(self, new_rules, filter, session: Optional[AsyncSession] = None, commit: bool = True) -> List[List[str]]: """_update_filtered_policies updates all the policies on the basis of the filter.""" - async with self._session_scope() as session: + async with self._session_scope(commit=commit, session=session) as s: # Load old policies stmt = select(self._db_class).where(self._db_class.ptype == filter.ptype) stmt = self._softdelete_query(stmt) filtered_stmt = self.filter_query(stmt, filter) - result = await session.execute(filtered_stmt) + result = await s.execute(filtered_stmt) old_rules_db = result.scalars().all() # Convert database objects to rule lists @@ -482,11 +545,11 @@ async def _update_filtered_policies(self, new_rules, filter) -> List[List[str]]: # Delete old policies - await self.remove_policies("p", filter.ptype, old_rules) + await self.remove_policies("p", filter.ptype, old_rules, session=s, commit=False) # Insert new policies - await self.add_policies("p", filter.ptype, new_rules) + await self.add_policies("p", filter.ptype, new_rules, session=s, commit=False) # return deleted rules diff --git a/tests/test_external_session.py b/tests/test_external_session.py index 4a85893..7c7bbc0 100644 --- a/tests/test_external_session.py +++ b/tests/test_external_session.py @@ -228,6 +228,241 @@ async def test_backward_compatibility(self): if os.path.exists(db_file.name): os.unlink(db_file.name) + # ------------------------------------------------------------------------- + # Helper for per-method session/commit tests + # ------------------------------------------------------------------------- + + async def _make_adapter(self): + """Create an in-memory SQLite engine, adapter, and session factory.""" + engine = create_async_engine("sqlite+aiosqlite://", future=True) + adapter = Adapter(engine) + await adapter.create_table() + session_factory = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) + return engine, adapter, session_factory + + async def _load_enforcer(self, engine): + """Create a fresh enforcer backed by the given engine.""" + e = casbin.AsyncEnforcer(get_fixture("rbac_model.conf"), Adapter(engine)) + await e.load_policy() + return e + + async def _make_enforcer(self, adapter): + """Create a fresh enforcer using an existing adapter instance.""" + e = casbin.AsyncEnforcer(get_fixture("rbac_model.conf"), adapter) + await e.load_policy() + return e + + async def test_per_method_session_commit(self): + """Test passing a session directly to adapter methods (commit path).""" + engine, adapter, session_factory = await self._make_adapter() + + async with session_factory() as session: + # add_policy with external session, no auto-commit + await adapter.add_policy("p", "p", ["alice", "data1", "read"], session=session, commit=False) + await adapter.add_policy("p", "p", ["bob", "data2", "write"], session=session, commit=False) + await session.commit() + + e = await self._load_enforcer(engine) + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + + async def test_per_method_session_rollback(self): + """Test passing a session directly to adapter methods (rollback path).""" + engine, adapter, session_factory = await self._make_adapter() + + async with session_factory() as session: + await adapter.add_policy("p", "p", ["alice", "data1", "read"], session=session, commit=False) + await session.rollback() + + e = await self._load_enforcer(engine) + self.assertFalse(e.enforce("alice", "data1", "read")) + + async def test_per_method_add_policies_session(self): + """Test add_policies with per-method session.""" + engine, adapter, session_factory = await self._make_adapter() + + async with session_factory() as session: + await adapter.add_policies("p", "p", [["alice", "data1", "read"], ["bob", "data2", "write"]], session=session, commit=False) + await session.commit() + + e = await self._load_enforcer(engine) + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + + async def test_per_method_remove_policy_session(self): + """Test remove_policy with per-method session.""" + engine, adapter, session_factory = await self._make_adapter() + + # Seed data without external session + await adapter.add_policy("p", "p", ["alice", "data1", "read"]) + + async with session_factory() as session: + result = await adapter.remove_policy("p", "p", ["alice", "data1", "read"], session=session, commit=False) + self.assertTrue(result) + await session.commit() + + e = await self._load_enforcer(engine) + self.assertFalse(e.enforce("alice", "data1", "read")) + + async def test_per_method_remove_filtered_policy_session(self): + """Test remove_filtered_policy with per-method session.""" + engine, adapter, session_factory = await self._make_adapter() + + await adapter.add_policies("p", "p", [["alice", "data1", "read"], ["alice", "data2", "read"]]) + + async with session_factory() as session: + result = await adapter.remove_filtered_policy("p", "p", 0, "alice", session=session, commit=False) + self.assertTrue(result) + await session.commit() + + e = await self._load_enforcer(engine) + self.assertFalse(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("alice", "data2", "read")) + + async def test_per_method_update_policy_session(self): + """Test update_policy with per-method session.""" + engine, adapter, session_factory = await self._make_adapter() + + await adapter.add_policy("p", "p", ["alice", "data1", "read"]) + + async with session_factory() as session: + await adapter.update_policy("p", "p", ["alice", "data1", "read"], ["alice", "data1", "write"], session=session, commit=False) + await session.commit() + + e = await self._load_enforcer(engine) + self.assertFalse(e.enforce("alice", "data1", "read")) + self.assertTrue(e.enforce("alice", "data1", "write")) + + async def test_per_method_update_policies_session(self): + """Test update_policies with per-method session.""" + engine, adapter, session_factory = await self._make_adapter() + + await adapter.add_policies("p", "p", [["alice", "data1", "read"], ["bob", "data2", "write"]]) + + async with session_factory() as session: + await adapter.update_policies( + "p", + "p", + [["alice", "data1", "read"], ["bob", "data2", "write"]], + [["alice", "data1", "write"], ["bob", "data2", "read"]], + session=session, + commit=False, + ) + await session.commit() + + e = await self._load_enforcer(engine) + self.assertTrue(e.enforce("alice", "data1", "write")) + self.assertTrue(e.enforce("bob", "data2", "read")) + self.assertFalse(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("bob", "data2", "write")) + + async def test_atomic_transaction_with_rollback(self): + """Test that multiple adapter operations in one session are atomic.""" + engine, adapter, session_factory = await self._make_adapter() + + async with session_factory() as session: + await adapter.add_policy("p", "p", ["alice", "data1", "read"], session=session, commit=False) + await adapter.add_policy("p", "p", ["bob", "data2", "write"], session=session, commit=False) + # Simulate an error by rolling back instead of committing + await session.rollback() + + # Neither policy should be persisted after rollback + e = await self._load_enforcer(engine) + self.assertFalse(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("bob", "data2", "write")) + + # ------------------------------------------------------------------------- + # Tests for the transaction() context manager (enforcer-friendly API) + # ------------------------------------------------------------------------- + + async def test_transaction_cm_commit(self): + """transaction() lets enforcer methods share a session; commit persists changes.""" + engine, adapter, session_factory = await self._make_adapter() + + # The enforcer MUST use the same adapter instance as the transaction() context. + e = await self._make_enforcer(adapter) + + async with session_factory() as session: + async with adapter.transaction(session=session): + await e.add_policy("alice", "data1", "read") + await e.add_policy("bob", "data2", "write") + # Changes are in the session but not committed yet + await session.commit() + + # Verify persistence + e2 = await self._load_enforcer(engine) + self.assertTrue(e2.enforce("alice", "data1", "read")) + self.assertTrue(e2.enforce("bob", "data2", "write")) + + async def test_transaction_cm_rollback(self): + """transaction() lets enforcer methods share a session; rollback discards changes.""" + engine, adapter, session_factory = await self._make_adapter() + + # The enforcer MUST use the same adapter instance as the transaction() context. + e = await self._make_enforcer(adapter) + + async with session_factory() as session: + async with adapter.transaction(session=session): + await e.add_policy("alice", "data1", "read") + await e.add_policy("bob", "data2", "write") + await session.rollback() + + # After rollback neither policy should exist + e2 = await self._load_enforcer(engine) + self.assertFalse(e2.enforce("alice", "data1", "read")) + self.assertFalse(e2.enforce("bob", "data2", "write")) + + async def test_transaction_cm_mixed_operations(self): + """transaction() works with add, remove, and update enforcer calls.""" + engine, adapter, session_factory = await self._make_adapter() + + # Seed some data first + await adapter.add_policy("p", "p", ["alice", "data1", "read"]) + await adapter.add_policy("p", "p", ["bob", "data2", "write"]) + + # The enforcer MUST use the same adapter instance as the transaction() context. + e = await self._make_enforcer(adapter) + + async with session_factory() as session: + async with adapter.transaction(session=session): + await e.remove_policy("alice", "data1", "read") + await e.add_policy("charlie", "data3", "read") + await session.commit() + + e2 = await self._load_enforcer(engine) + self.assertFalse(e2.enforce("alice", "data1", "read")) + self.assertTrue(e2.enforce("bob", "data2", "write")) + self.assertTrue(e2.enforce("charlie", "data3", "read")) + + async def test_transaction_cm_does_not_affect_other_tasks(self): + """transaction() ContextVar is isolated to the current asyncio task.""" + import asyncio + + engine, adapter, session_factory = await self._make_adapter() + + e = await self._load_enforcer(engine) + + results = {} + + async def task_with_transaction(): + async with session_factory() as session: + async with adapter.transaction(session=session): + # Peek at the tx session from inside the task + from casbin_async_sqlalchemy_adapter.adapter import _UNSET + + results["inside"] = adapter._tx_session_var.get() is not _UNSET + await session.rollback() + + async def task_without_transaction(): + from casbin_async_sqlalchemy_adapter.adapter import _UNSET + + results["outside"] = adapter._tx_session_var.get() is _UNSET + + await asyncio.gather(task_with_transaction(), task_without_transaction()) + + self.assertTrue(results["inside"]) + self.assertTrue(results["outside"]) + if __name__ == "__main__": unittest.main()