If your application uses async/await and relies heavily on I/O operations, use casbin.AsyncEnforcer.
To use it:
- Initialize an async engine and an async adapter (a subclass of
AsyncAdapter). - Create the
AsyncEnforcer instance by passing the model file path and the async adapter. - Call
await e.load_policy() to load the policies from the adapter. - Use
await e.enforce(...) to check permissions.
Built-in async adapters are available in casbin.persist.adapters.asyncio.
import asyncio
import casbin
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from casbin_async_sqlalchemy_adapter import Adapter, CasbinRule
async def get_enforcer():
engine = create_async_engine("sqlite+aiosqlite://", future=True)
adapter = Adapter(engine)
await adapter.create_table()
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
async with async_session() as s:
s.add(CasbinRule(ptype="p", v0="alice", v1="data1", v2="read"))
s.add(CasbinRule(ptype="p", v0="bob", v1="data2", v2="write"))
s.add(CasbinRule(ptype="p", v0="data2_admin", v1="data2", v2="read"))
s.add(CasbinRule(ptype="p", v0="data2_admin", v1="data2", v2="write"))
s.add(CasbinRule(ptype="g", v0="alice", v1="data2_admin"))
await s.commit()
e = casbin.AsyncEnforcer("path/to/model.conf", adapter)
await e.load_policy()
return e
async def main():
e = await get_enforcer()
if e.enforce("alice", "data1", "read"):
print("alice can read data1")
else:
print("alice can not read data1")
asyncio.run(main())
