casbin-sqlalchemy-adapter

import casbin from casbin_sqlalchemy_adapter import Adapter import os # Define paths for your Casbin model and policy files # For a real application, these would be proper paths to .conf and .csv files # Example 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 # # Example policy.csv content: # p, alice, data1, read # p, bob, data2, write model_conf_path = os.path.join(os.path.dirname(__file__), "model.conf") policy_csv_path = os.path.join(os.path.dirname(__file__), "policy.csv") # Create dummy model and policy files for demonstration if they don't exist if not os.path.exists(model_conf_path): with open(model_conf_path, 'w') as f: f.write(""" [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 """) if not os.path.exists(policy_csv_path): with open(policy_csv_path, 'w') as f: f.write(""" p, alice, data1, read p, bob, data2, write """) # Initialize the SQLAlchemy adapter with a SQLite database # Replace 'sqlite:///test.db' with your database connection string (e.g., 'mysql+pymysql://user:password@host/dbname') adapter = Adapter('sqlite:///test.db') # # Initialize the Casbin Enforcer # The enforcer will load the model from model.conf and policies from the adapter e = casbin.Enforcer(model_conf_path, adapter) # # Add a policy via the Enforcer (this will be saved to the database) # Note: policies added directly via the Enforcer API are automatically saved if Auto-Save is enabled (default). if e.add_policy("charlie", "data3", "edit"): print("Policy 'charlie, data3, edit' added.") # Enforce a policy sub = "alice" obj = "data1" act = "read" if e.enforce(sub, obj, act): print(f"{sub} is permitted to {act} {obj}.") else: print(f"{sub} is denied to {act} {obj}.") sub = "bob" obj = "data1" act = "read" if e.enforce(sub, obj, act): print(f"{sub} is permitted to {act} {obj}.") else: print(f"{sub} is denied to {act} {obj}.") sub = "charlie" obj = "data3" act = "edit" if e.enforce(sub, obj, act): print(f"{sub} is permitted to {act} {obj}.") else: print(f"{sub} is denied to {act} {obj}.") # Remove a policy if e.remove_policy("charlie", "data3", "edit"): print("Policy 'charlie, data3, edit' removed.") # Check again after removal if not e.enforce("charlie", "data3", "edit"): print("Charlie is now denied to edit data3 as expected.")