From 671f3a8dec73cf3600ed70f3ac69962c44f747e6 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 2 Dec 2025 12:46:54 +0000 Subject: [PATCH 1/3] Initial plan From b3330ebb38625c955009c9fda2e82a5129c89dee Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 2 Dec 2025 12:54:23 +0000 Subject: [PATCH 2/3] Add create_casbin_rule_model function for Alembic migration support Co-authored-by: hsluoyz <3787410+hsluoyz@users.noreply.github.com> --- README.md | 75 +++++++++++++++++++ casbin_async_sqlalchemy_adapter/__init__.py | 2 +- casbin_async_sqlalchemy_adapter/adapter.py | 55 ++++++++++++++ tests/test_adapter.py | 81 +++++++++++++++++++++ 4 files changed, 212 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 498c3fe..537a73c 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,81 @@ When soft deletion is enabled: This feature maintains full backward compatibility - when `db_class_softdelete_attribute` is not provided, the adapter functions with hard deletion as before. +## Alembic Migration Support + +The adapter supports integration with Alembic for database migrations. Instead of using `adapter.create_table()` at application startup, you can use Alembic to manage the `casbin_rule` table alongside your other application tables. + +### Using Your Application's Base + +Use the `create_casbin_rule_model()` function to create a `CasbinRule` model that uses your application's existing SQLAlchemy `Base`: + +```python +# In your models.py or wherever you define your SQLAlchemy models +from sqlalchemy.orm import declarative_base +from casbin_async_sqlalchemy_adapter import create_casbin_rule_model + +# Your application's Base +Base = declarative_base() + +# Create CasbinRule model using your Base +CasbinRule = create_casbin_rule_model(Base) + +# Define your other models using the same Base +class User(Base): + __tablename__ = "users" + # ... your columns +``` + +### Setting Up Alembic + +1. Make sure your `CasbinRule` model is imported in your Alembic `env.py`: + +```python +# In alembic/env.py +from myapp.models import Base # Import your Base that includes CasbinRule + +target_metadata = Base.metadata +``` + +2. Generate the migration: + +```bash +alembic revision --autogenerate -m "add casbin_rule table" +``` + +3. Apply the migration: + +```bash +alembic upgrade head +``` + +### Using the Custom Model with the Adapter + +When using a custom CasbinRule model created with `create_casbin_rule_model()`, pass it to the Adapter: + +```python +from sqlalchemy.ext.asyncio import create_async_engine +from casbin_async_sqlalchemy_adapter import Adapter, create_casbin_rule_model +from myapp.models import Base + +# Create CasbinRule using your Base +CasbinRule = create_casbin_rule_model(Base) + +engine = create_async_engine('sqlite+aiosqlite:///test.db') +adapter = Adapter(engine, db_class=CasbinRule) + +# No need to call adapter.create_table() - Alembic handles the migration +e = casbin.AsyncEnforcer('path/to/model.conf', adapter) +``` + +### Custom Table Name + +You can also specify a custom table name: + +```python +CasbinRule = create_casbin_rule_model(Base, table_name="my_casbin_rules") +``` + ### Getting Help - [PyCasbin](https://github.com/casbin/pycasbin) diff --git a/casbin_async_sqlalchemy_adapter/__init__.py b/casbin_async_sqlalchemy_adapter/__init__.py index e66fe6e..e5b3515 100644 --- a/casbin_async_sqlalchemy_adapter/__init__.py +++ b/casbin_async_sqlalchemy_adapter/__init__.py @@ -12,4 +12,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -from .adapter import CasbinRule, Adapter, Base +from .adapter import CasbinRule, Adapter, Base, create_casbin_rule_model diff --git a/casbin_async_sqlalchemy_adapter/adapter.py b/casbin_async_sqlalchemy_adapter/adapter.py index 4af3e7c..9adb539 100644 --- a/casbin_async_sqlalchemy_adapter/adapter.py +++ b/casbin_async_sqlalchemy_adapter/adapter.py @@ -49,6 +49,61 @@ def __repr__(self): return ''.format(self.id, str(self)) +def create_casbin_rule_model(base, table_name="casbin_rule"): + """ + Create a CasbinRule model class using the given SQLAlchemy declarative base. + + This function allows you to create a CasbinRule model that uses your application's + existing Base metadata, enabling integration with Alembic migrations. + + Args: + base: The SQLAlchemy declarative base class to use for the model. + table_name: The name of the table to use. Defaults to "casbin_rule". + + Returns: + A CasbinRule model class that uses the given base's metadata. + + Example: + from sqlalchemy.orm import declarative_base + from casbin_async_sqlalchemy_adapter import Adapter, create_casbin_rule_model + + # Use your application's existing Base + Base = declarative_base() + + # Create CasbinRule using your Base + CasbinRule = create_casbin_rule_model(Base) + + # Now CasbinRule will be included in Alembic auto-generated migrations + # when you run: alembic revision --autogenerate + """ + + class CasbinRuleModel(base): + __tablename__ = table_name + __table_args__ = {"extend_existing": True} + + id = Column(Integer, primary_key=True) + ptype = Column(String(255)) + v0 = Column(String(255)) + v1 = Column(String(255)) + v2 = Column(String(255)) + v3 = Column(String(255)) + v4 = Column(String(255)) + v5 = Column(String(255)) + + def __str__(self): + arr = [self.ptype] + for v in (self.v0, self.v1, self.v2, self.v3, self.v4, self.v5): + if v is None: + break + arr.append(v) + return ", ".join(arr) + + def __repr__(self): + return ''.format(self.id, str(self)) + + return CasbinRuleModel + + class Filter: ptype = [] v0 = [] diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 72e2f09..546b6b6 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -19,10 +19,12 @@ import casbin from sqlalchemy import Column, Integer, String, select from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker +from sqlalchemy.orm import declarative_base from casbin_async_sqlalchemy_adapter import Adapter from casbin_async_sqlalchemy_adapter import Base from casbin_async_sqlalchemy_adapter import CasbinRule +from casbin_async_sqlalchemy_adapter import create_casbin_rule_model from casbin_async_sqlalchemy_adapter.adapter import Filter @@ -418,5 +420,84 @@ async def test_add_policies_bulk_internal_session(self): assert r in tuples +class TestCreateCasbinRuleModel(IsolatedAsyncioTestCase): + async def test_create_casbin_rule_model_with_custom_base(self): + """Test that create_casbin_rule_model creates a model with the given base's metadata.""" + # Create a custom Base (simulating user's application Base) + CustomBase = declarative_base() + + # Use a unique table name to avoid conflicts with other tests + CustomCasbinRule = create_casbin_rule_model(CustomBase, table_name="test_custom_rule") + + # Verify the model uses the custom Base's metadata + self.assertIn("test_custom_rule", CustomBase.metadata.tables) + + # Verify the model has all required columns + self.assertTrue(hasattr(CustomCasbinRule, "id")) + self.assertTrue(hasattr(CustomCasbinRule, "ptype")) + self.assertTrue(hasattr(CustomCasbinRule, "v0")) + self.assertTrue(hasattr(CustomCasbinRule, "v1")) + self.assertTrue(hasattr(CustomCasbinRule, "v2")) + self.assertTrue(hasattr(CustomCasbinRule, "v3")) + self.assertTrue(hasattr(CustomCasbinRule, "v4")) + self.assertTrue(hasattr(CustomCasbinRule, "v5")) + + # Create engine and table + engine = create_async_engine("sqlite+aiosqlite://", future=True) + async with engine.begin() as conn: + await conn.run_sync(CustomBase.metadata.create_all) + + # Test that we can insert and query records + async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) + async with async_session() as s: + s.add(CustomCasbinRule(ptype="p", v0="alice", v1="data1", v2="read")) + await s.commit() + result = await s.execute(select(CustomCasbinRule)) + rules = result.scalars().all() + self.assertEqual(len(rules), 1) + self.assertEqual(str(rules[0]), "p, alice, data1, read") + + async def test_create_casbin_rule_model_with_custom_table_name(self): + """Test that create_casbin_rule_model respects custom table name.""" + CustomBase = declarative_base() + CustomCasbinRule = create_casbin_rule_model(CustomBase, table_name="my_custom_rules") + + self.assertIn("my_custom_rules", CustomBase.metadata.tables) + self.assertEqual(CustomCasbinRule.__tablename__, "my_custom_rules") + + async def test_create_casbin_rule_model_str_and_repr(self): + """Test that str and repr methods work correctly on the created model.""" + CustomBase = declarative_base() + CustomCasbinRule = create_casbin_rule_model(CustomBase, table_name="str_repr_test_rules") + + rule = CustomCasbinRule(ptype="p", v0="alice", v1="data1", v2="read") + self.assertEqual(str(rule), "p, alice, data1, read") + self.assertEqual(repr(rule), '') + + async def test_create_casbin_rule_model_with_adapter(self): + """Test that the created model works with the Adapter (uses adapter's Base to avoid side effects).""" + # Use the adapter's Base to create the rule model - this avoids the side effect + # of Base.metadata = db_class.metadata in the Adapter constructor + CustomCasbinRule = create_casbin_rule_model(Base, table_name="adapter_test_rule") + + # Verify the model uses Base's metadata + self.assertIn("adapter_test_rule", Base.metadata.tables) + + # Create engine and table + engine = create_async_engine("sqlite+aiosqlite://", future=True) + await Adapter(engine, db_class=CustomCasbinRule).create_table() + + async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) + async with async_session() as s: + s.add(CustomCasbinRule(ptype="p", v0="bob", v1="data2", v2="write")) + await s.commit() + + # Test that the model works with the Adapter + adapter = Adapter(engine, db_class=CustomCasbinRule) + e = casbin.AsyncEnforcer(get_fixture("rbac_model.conf"), adapter) + await e.load_policy() + self.assertTrue(e.enforce("bob", "data2", "write")) + + if __name__ == "__main__": unittest.main() From 9f9c83c33c5c4729495c10961c27bb296ae987e7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 7 Dec 2025 13:13:23 +0000 Subject: [PATCH 3/3] Simplify: remove README docs and tests, shorten docstring Co-authored-by: hsluoyz <3787410+hsluoyz@users.noreply.github.com> --- README.md | 75 -------------------- casbin_async_sqlalchemy_adapter/adapter.py | 27 +------- tests/test_adapter.py | 81 ---------------------- 3 files changed, 1 insertion(+), 182 deletions(-) diff --git a/README.md b/README.md index 537a73c..498c3fe 100644 --- a/README.md +++ b/README.md @@ -202,81 +202,6 @@ When soft deletion is enabled: This feature maintains full backward compatibility - when `db_class_softdelete_attribute` is not provided, the adapter functions with hard deletion as before. -## Alembic Migration Support - -The adapter supports integration with Alembic for database migrations. Instead of using `adapter.create_table()` at application startup, you can use Alembic to manage the `casbin_rule` table alongside your other application tables. - -### Using Your Application's Base - -Use the `create_casbin_rule_model()` function to create a `CasbinRule` model that uses your application's existing SQLAlchemy `Base`: - -```python -# In your models.py or wherever you define your SQLAlchemy models -from sqlalchemy.orm import declarative_base -from casbin_async_sqlalchemy_adapter import create_casbin_rule_model - -# Your application's Base -Base = declarative_base() - -# Create CasbinRule model using your Base -CasbinRule = create_casbin_rule_model(Base) - -# Define your other models using the same Base -class User(Base): - __tablename__ = "users" - # ... your columns -``` - -### Setting Up Alembic - -1. Make sure your `CasbinRule` model is imported in your Alembic `env.py`: - -```python -# In alembic/env.py -from myapp.models import Base # Import your Base that includes CasbinRule - -target_metadata = Base.metadata -``` - -2. Generate the migration: - -```bash -alembic revision --autogenerate -m "add casbin_rule table" -``` - -3. Apply the migration: - -```bash -alembic upgrade head -``` - -### Using the Custom Model with the Adapter - -When using a custom CasbinRule model created with `create_casbin_rule_model()`, pass it to the Adapter: - -```python -from sqlalchemy.ext.asyncio import create_async_engine -from casbin_async_sqlalchemy_adapter import Adapter, create_casbin_rule_model -from myapp.models import Base - -# Create CasbinRule using your Base -CasbinRule = create_casbin_rule_model(Base) - -engine = create_async_engine('sqlite+aiosqlite:///test.db') -adapter = Adapter(engine, db_class=CasbinRule) - -# No need to call adapter.create_table() - Alembic handles the migration -e = casbin.AsyncEnforcer('path/to/model.conf', adapter) -``` - -### Custom Table Name - -You can also specify a custom table name: - -```python -CasbinRule = create_casbin_rule_model(Base, table_name="my_casbin_rules") -``` - ### Getting Help - [PyCasbin](https://github.com/casbin/pycasbin) diff --git a/casbin_async_sqlalchemy_adapter/adapter.py b/casbin_async_sqlalchemy_adapter/adapter.py index 9adb539..66a8286 100644 --- a/casbin_async_sqlalchemy_adapter/adapter.py +++ b/casbin_async_sqlalchemy_adapter/adapter.py @@ -50,32 +50,7 @@ def __repr__(self): def create_casbin_rule_model(base, table_name="casbin_rule"): - """ - Create a CasbinRule model class using the given SQLAlchemy declarative base. - - This function allows you to create a CasbinRule model that uses your application's - existing Base metadata, enabling integration with Alembic migrations. - - Args: - base: The SQLAlchemy declarative base class to use for the model. - table_name: The name of the table to use. Defaults to "casbin_rule". - - Returns: - A CasbinRule model class that uses the given base's metadata. - - Example: - from sqlalchemy.orm import declarative_base - from casbin_async_sqlalchemy_adapter import Adapter, create_casbin_rule_model - - # Use your application's existing Base - Base = declarative_base() - - # Create CasbinRule using your Base - CasbinRule = create_casbin_rule_model(Base) - - # Now CasbinRule will be included in Alembic auto-generated migrations - # when you run: alembic revision --autogenerate - """ + """Create a CasbinRule model using the given declarative base for Alembic integration.""" class CasbinRuleModel(base): __tablename__ = table_name diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 546b6b6..72e2f09 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -19,12 +19,10 @@ import casbin from sqlalchemy import Column, Integer, String, select from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker -from sqlalchemy.orm import declarative_base from casbin_async_sqlalchemy_adapter import Adapter from casbin_async_sqlalchemy_adapter import Base from casbin_async_sqlalchemy_adapter import CasbinRule -from casbin_async_sqlalchemy_adapter import create_casbin_rule_model from casbin_async_sqlalchemy_adapter.adapter import Filter @@ -420,84 +418,5 @@ async def test_add_policies_bulk_internal_session(self): assert r in tuples -class TestCreateCasbinRuleModel(IsolatedAsyncioTestCase): - async def test_create_casbin_rule_model_with_custom_base(self): - """Test that create_casbin_rule_model creates a model with the given base's metadata.""" - # Create a custom Base (simulating user's application Base) - CustomBase = declarative_base() - - # Use a unique table name to avoid conflicts with other tests - CustomCasbinRule = create_casbin_rule_model(CustomBase, table_name="test_custom_rule") - - # Verify the model uses the custom Base's metadata - self.assertIn("test_custom_rule", CustomBase.metadata.tables) - - # Verify the model has all required columns - self.assertTrue(hasattr(CustomCasbinRule, "id")) - self.assertTrue(hasattr(CustomCasbinRule, "ptype")) - self.assertTrue(hasattr(CustomCasbinRule, "v0")) - self.assertTrue(hasattr(CustomCasbinRule, "v1")) - self.assertTrue(hasattr(CustomCasbinRule, "v2")) - self.assertTrue(hasattr(CustomCasbinRule, "v3")) - self.assertTrue(hasattr(CustomCasbinRule, "v4")) - self.assertTrue(hasattr(CustomCasbinRule, "v5")) - - # Create engine and table - engine = create_async_engine("sqlite+aiosqlite://", future=True) - async with engine.begin() as conn: - await conn.run_sync(CustomBase.metadata.create_all) - - # Test that we can insert and query records - async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) - async with async_session() as s: - s.add(CustomCasbinRule(ptype="p", v0="alice", v1="data1", v2="read")) - await s.commit() - result = await s.execute(select(CustomCasbinRule)) - rules = result.scalars().all() - self.assertEqual(len(rules), 1) - self.assertEqual(str(rules[0]), "p, alice, data1, read") - - async def test_create_casbin_rule_model_with_custom_table_name(self): - """Test that create_casbin_rule_model respects custom table name.""" - CustomBase = declarative_base() - CustomCasbinRule = create_casbin_rule_model(CustomBase, table_name="my_custom_rules") - - self.assertIn("my_custom_rules", CustomBase.metadata.tables) - self.assertEqual(CustomCasbinRule.__tablename__, "my_custom_rules") - - async def test_create_casbin_rule_model_str_and_repr(self): - """Test that str and repr methods work correctly on the created model.""" - CustomBase = declarative_base() - CustomCasbinRule = create_casbin_rule_model(CustomBase, table_name="str_repr_test_rules") - - rule = CustomCasbinRule(ptype="p", v0="alice", v1="data1", v2="read") - self.assertEqual(str(rule), "p, alice, data1, read") - self.assertEqual(repr(rule), '') - - async def test_create_casbin_rule_model_with_adapter(self): - """Test that the created model works with the Adapter (uses adapter's Base to avoid side effects).""" - # Use the adapter's Base to create the rule model - this avoids the side effect - # of Base.metadata = db_class.metadata in the Adapter constructor - CustomCasbinRule = create_casbin_rule_model(Base, table_name="adapter_test_rule") - - # Verify the model uses Base's metadata - self.assertIn("adapter_test_rule", Base.metadata.tables) - - # Create engine and table - engine = create_async_engine("sqlite+aiosqlite://", future=True) - await Adapter(engine, db_class=CustomCasbinRule).create_table() - - async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) - async with async_session() as s: - s.add(CustomCasbinRule(ptype="p", v0="bob", v1="data2", v2="write")) - await s.commit() - - # Test that the model works with the Adapter - adapter = Adapter(engine, db_class=CustomCasbinRule) - e = casbin.AsyncEnforcer(get_fixture("rbac_model.conf"), adapter) - await e.load_policy() - self.assertTrue(e.enforce("bob", "data2", "write")) - - if __name__ == "__main__": unittest.main()