sqlalchemy-adapter

import os import casbin import tempfile from sqlalchemy_adapter import Adapter # Create a dummy model.conf for the Enforcer model_conf_content = ''' [request_definition] r = sub, obj, act [policy_definition] p = sub, obj, act [policy_effect] e = some(where (p.eft == allow)) [matchers] m = r.sub == p.sub && r.obj == p.obj && r.act == p.act ''' # Write the model content to a temporary file with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix=".conf") as model_file: model_file.write(model_conf_content) model_path = model_file.name # Use an in-memory SQLite database for a simple, runnable example. # For persistent storage, provide a file path or a full database URL # e.g., 'sqlite:///test.db' or from an environment variable. database_url = os.environ.get('SQLALCHEMY_DATABASE_URL', 'sqlite:///:memory:') try: # 1. Initialize the SQLAlchemy Adapter # By default, create_table=True, so tables will be created if they don't exist. adapter = Adapter(database_url) # 2. Initialize Casbin Enforcer with the adapter e = casbin.Enforcer(model_path, adapter) # 3. Add policies e.add_policy("alice", "data1", "read") e.add_policy("bob", "data2", "write") # 4. Save policies to the database (persists changes through the adapter) e.save_policy() print("Policies added and saved.") # 5. Enforce policies print(f"Alice can read data1: {e.enforce('alice', 'data1', 'read')}") # Expected: True print(f"Bob can write data2: {e.enforce('bob', 'data2', 'write')}") # Expected: True print(f"Alice can write data1: {e.enforce('alice', 'data1', 'write')}") # Expected: False # 6. Load policies (useful if Enforcer was re-initialized or policy was changed elsewhere) e.load_policy() print(f"After reloading, Alice can read data1: {e.enforce('alice', 'data1', 'read')}") # Expected: True except Exception as ex: print(f"An error occurred: {ex}") finally: # Clean up the temporary model file os.remove(model_path)