Skip to content

feat: add clear_policy() method to Adapter #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mserico merged 7 commits into from
Dec 10, 2025
Merged

feat: add clear_policy() method to Adapter #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
1bb3d3f
Initial plan
Copilot Dec 10, 2025
0bb2678
Add clear_policy() method to Adapter for database cleanup
Copilot Dec 10, 2025
284f1d8
Move func import to top of file for better organization
Copilot Dec 10, 2025
c03e468
Complete implementation of clear_policy() method
Copilot Dec 10, 2025
f9c4add
Remove example.db and add *.db to .gitignore
Copilot Dec 10, 2025
8fa57a8
Fix Black linter error - remove trailing whitespace in docstring
Copilot Dec 10, 2025
adf8405
Add clear_policy() API documentation to README
Copilot Dec 10, 2025
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
4 changes: 4 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,3 +130,7 @@ dmypy.json

.idea/
*.iml

# SQLite database files
*.db
*.db-journal
30 changes: 30 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,36 @@ async with async_session() as session:
await session.commit()
```

## Clearing All Policies

The adapter provides a `clear_policy()` method to remove all policy records from the database directly:

```python
import casbin_async_sqlalchemy_adapter
import casbin
from sqlalchemy.ext.asyncio import create_async_engine

# Setup
engine = create_async_engine('sqlite+aiosqlite:///test.db')
adapter = casbin_async_sqlalchemy_adapter.Adapter(engine)
await adapter.create_table()

e = casbin.AsyncEnforcer('path/to/model.conf', adapter)
await e.load_policy()

# Add some policies
await e.add_policy("alice", "data1", "read")
await e.add_policy("bob", "data2", "write")

# Clear all policies from the database
await adapter.clear_policy()

# Reload to verify - the enforcer will have no policies
await e.load_policy()
```

When soft deletion is enabled, `clear_policy()` marks all records as deleted instead of physically removing them.

## Soft Deletion Support

The adapter supports soft deletion, which marks records as deleted instead of physically removing them from the database. This is useful for:
Expand Down
24 changes: 24 additions & 0 deletions casbin_async_sqlalchemy_adapter/adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,30 @@ async def save_policy(self, model):

return True

async def clear_policy(self):
"""Clears all policy rules from the storage (database).

This method removes all records from the casbin_rule table.
If soft delete is enabled, it marks all records as deleted.

Returns:
bool: True if successful, False otherwise.
"""
async with self._session_scope() as session:
if self.softdelete_attribute is None:
# Hard delete all records
stmt = delete(self._db_class)
await session.execute(stmt)
else:
# Soft delete all active records
stmt = select(self._db_class)
stmt = self._softdelete_query(stmt)
result = await session.execute(stmt)
lines = result.scalars().all()
for line in lines:
setattr(line, self.softdelete_attribute.name, True)
return True

async def add_policy(self, sec, ptype, rule):
"""adds a policy rule to the storage."""
await self._save_policy_line(ptype, rule)
Expand Down
32 changes: 31 additions & 1 deletion tests/test_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
from unittest import IsolatedAsyncioTestCase

import casbin
from sqlalchemy import Column, Integer, String, select
from sqlalchemy import Column, Integer, String, select, func
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

from casbin_async_sqlalchemy_adapter import Adapter
Expand Down Expand Up @@ -390,6 +390,36 @@ async def test_update_filtered_policies(self):
await e.update_filtered_policies([["bob", "data2", "read"]], 0, "bob")
self.assertTrue(e.enforce("bob", "data2", "read"))

async def test_clear_policy(self):
"""Test that clear_policy() removes all records from the database."""
e = await get_enforcer()
adapter = e.get_adapter()
engine = adapter._engine

# Verify there are policies in the database
async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
async with async_session() as s:
cnt = await s.execute(select(func.count()).select_from(CasbinRule))
initial_count = cnt.scalar_one()
self.assertGreater(initial_count, 0, "There should be policies in the database before clearing")

# Clear all policies from the database
await adapter.clear_policy()

# Verify all policies are removed from the database
async with async_session() as s:
cnt = await s.execute(select(func.count()).select_from(CasbinRule))
final_count = cnt.scalar_one()
self.assertEqual(final_count, 0, "All policies should be removed from the database")

# Verify enforcer still works after clearing (can load empty policy)
await e.load_policy()
self.assertFalse(e.enforce("alice", "data1", "read"))

# Verify we can add policies after clearing
await e.add_policy("eve", "data3", "read")
self.assertTrue(e.enforce("eve", "data3", "read"))


class TestBulkInsert(IsolatedAsyncioTestCase):
async def test_add_policies_bulk_internal_session(self):
Expand Down
47 changes: 47 additions & 0 deletions tests/test_adapter_softdelete.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,3 +342,50 @@ async def test_load_filtered_policy_ignores_soft_deleted(self):
self.assertFalse(e2.enforce("bob", "data2", "write"))
# Other data2 policies should be loaded
self.assertTrue(e2.enforce("data2_admin", "data2", "read"))

async def test_clear_policy_with_softdelete(self):
"""Test that clear_policy() marks all records as deleted when softdelete is enabled."""
e = await self.get_enforcer()
adapter = e.get_adapter()
engine = adapter._engine

# Verify there are policies in the database
async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
async with async_session() as s:
# Count total records (including soft-deleted)
total_result = await s.execute(select(CasbinRuleSoftDelete))
total_count = len(total_result.scalars().all())
self.assertGreater(total_count, 0, "There should be policies in the database before clearing")

# Count non-deleted records
active_result = await s.execute(select(CasbinRuleSoftDelete).where(CasbinRuleSoftDelete.is_deleted == False))
active_count = len(active_result.scalars().all())
self.assertGreater(active_count, 0, "There should be active policies before clearing")

# Clear all policies (soft delete)
await adapter.clear_policy()

# Verify all active policies are now marked as deleted
async with async_session() as s:
# Total count should remain the same (soft delete)
total_result = await s.execute(select(CasbinRuleSoftDelete))
total_after = len(total_result.scalars().all())
self.assertEqual(total_count, total_after, "Total records should remain the same with soft delete")

# Active count should be 0
active_result = await s.execute(select(CasbinRuleSoftDelete).where(CasbinRuleSoftDelete.is_deleted == False))
active_after = len(active_result.scalars().all())
self.assertEqual(active_after, 0, "All policies should be marked as deleted")

# All should be marked as deleted
deleted_result = await s.execute(select(CasbinRuleSoftDelete).where(CasbinRuleSoftDelete.is_deleted == True))
deleted_after = len(deleted_result.scalars().all())
self.assertEqual(deleted_after, total_count, "All policies should be marked as deleted")

# Verify enforcer still works after clearing (can load empty policy)
await e.load_policy()
self.assertFalse(e.enforce("alice", "data1", "read"))

# Verify we can add policies after clearing
await e.add_policy("eve", "data3", "read")
self.assertTrue(e.enforce("eve", "data3", "read"))