Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions casbin_async_sqlalchemy_adapter/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

from casbin import persist
from casbin.persist.adapters.asyncio import AsyncAdapter
from sqlalchemy import Column, Integer, String, delete
from sqlalchemy import Column, Integer, String, delete, insert
from sqlalchemy import or_
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.future import select
Expand Down Expand Up @@ -190,14 +190,20 @@ async def add_policy(self, sec, ptype, rule):

async def add_policies(self, sec, ptype, rules):
"""adds a policy rules to the storage."""
if self._external_session is not None:
# Use external session to add all rules in the same transaction
for rule in rules:
await self._save_policy_line(ptype, rule, self._external_session)
else:
# Use individual sessions for each rule (original behavior)
for rule in rules:
await self._save_policy_line(ptype, rule)
if not rules:
return

# Build rows for executemany bulk insert
rows = []
for rule in rules:
row = {"ptype": ptype}
for i, v in enumerate(rule):
row[f"v{i}"] = v
rows.append(row)

async with self._session_scope() as session:
stmt = insert(self._db_class)
await session.execute(stmt, rows)

async def remove_policy(self, sec, ptype, rule):
"""removes a policy rule from the storage."""
Expand Down
37 changes: 37 additions & 0 deletions tests/test_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -399,5 +399,42 @@ async def test_update_filtered_policies(self):
self.assertTrue(e.enforce("bob", "data2", "read"))


class TestBulkInsert(IsolatedAsyncioTestCase):
async def test_add_policies_bulk_internal_session(self):
engine = create_async_engine("sqlite+aiosqlite://", future=True)
adapter = Adapter(engine)
await adapter.create_table()

rules = [
("u1", "obj1", "read"),
("u2", "obj2", "write"),
("u3", "obj3", "read"),
]
await adapter.add_policies("p", "p", rules)

async_session = async_sessionmaker(
engine, expire_on_commit=False, class_=AsyncSession
)
async with async_session() as s:
# count inserted rows
from sqlalchemy import select, func

cnt = await s.execute(
select(func.count())
.select_from(CasbinRule)
.where(CasbinRule.ptype == "p")
)
assert cnt.scalar_one() == len(rules)

rows = (
(await s.execute(select(CasbinRule).order_by(CasbinRule.id)))
.scalars()
.all()
)
tuples = [(r.v0, r.v1, r.v2) for r in rows]
for r in rules:
assert r in tuples


if __name__ == "__main__":
unittest.main()