Database Example

A complete example using casbin-fastapi-decorator with the db extra. Policies are stored in SQLite, and the cached enforcer hot-reloads when the database or model.conf changes.

Install

pip install "casbin-fastapi-decorator[db]" uvicorn aiosqlite python-multipart

Project structure

my-app/
├── casbin/
│   └── model.conf
├── auth.py
├── authz.py
├── db.py
├── model.py
└── main.py

No policy.csv needed — policies live in the database.

casbin/model.conf

Same as the basic example.

model.py

from enum import StrEnum
from pydantic import BaseModel

class UserSchema(BaseModel):
    role: str

class PostCreateSchema(BaseModel):
    title: str

class PostSchema(PostCreateSchema):
    id: int

class Permission(StrEnum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"

class Resource(StrEnum):
    POST = "post"

class Role(StrEnum):
    ADMIN = "admin"
    EDITOR = "editor"
    VIEWER = "viewer"

db.py

from model import Permission, Resource, Role
from sqlalchemy import String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

engine = create_async_engine("sqlite+aiosqlite:///./example.db")
async_session = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

class Base(DeclarativeBase):
    pass

class Policy(Base):
    __tablename__ = "policies"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    sub: Mapped[str] = mapped_column(String(100))
    obj: Mapped[str] = mapped_column(String(100))
    act: Mapped[str] = mapped_column(String(100))

async def seed_policies() -> None:
    async with async_session() as session:
        result = await session.execute(select(Policy))
        if result.scalars().first() is not None:
            return
        session.add_all([
            Policy(sub=Role.ADMIN,   obj=Resource.POST, act=Permission.READ),
            Policy(sub=Role.ADMIN,   obj=Resource.POST, act=Permission.WRITE),
            Policy(sub=Role.ADMIN,   obj=Resource.POST, act=Permission.DELETE),
            Policy(sub=Role.EDITOR,  obj=Resource.POST, act=Permission.READ),
            Policy(sub=Role.EDITOR,  obj=Resource.POST, act=Permission.WRITE),
            Policy(sub=Role.VIEWER,  obj=Resource.POST, act=Permission.READ),
        ])
        await session.commit()

async def setup_db() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    await seed_policies()

auth.py

from typing import Annotated
from fastapi import HTTPException, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from model import UserSchema

async def get_current_user(
    header_auth: Annotated[
        HTTPAuthorizationCredentials | None,
        Security(HTTPBearer(auto_error=False))
    ],
) -> UserSchema:
    if not header_auth:
        raise HTTPException(401, "Unauthorized")
    return UserSchema(role=header_auth.credentials)

authz.py

from auth import get_current_user
from casbin_fastapi_decorator_db import DatabaseEnforcerProvider
from db import Policy, async_session
from fastapi import HTTPException

from casbin_fastapi_decorator import PermissionGuard

enforcer_provider = DatabaseEnforcerProvider(
    model_path="casbin/model.conf",
    session_factory=async_session,
    policy_model=Policy,
    policy_mapper=lambda p: (p.sub, p.obj, p.act),
)

guard = PermissionGuard(
    user_provider=get_current_user,
    enforcer_provider=enforcer_provider,
    error_factory=lambda *_: HTTPException(403, "Forbidden"),
)

main.py

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated
from auth import get_current_user
from authz import enforcer_provider, guard
from db import Policy, async_session, engine, setup_db
from fastapi import Depends, FastAPI, Form
from model import Permission, PostCreateSchema, PostSchema, Resource, Role, UserSchema
from sqlalchemy import select

@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
    await setup_db()
    async with enforcer_provider:
        yield
    await engine.dispose()

app = FastAPI(title="Core + DB Example", lifespan=lifespan)

MOCK_DB = [
    PostSchema(id=1, title="First Post"),
    PostSchema(id=2, title="Second Post"),
]

@app.post("/login")
async def login(role: Role) -> str:
    return role

@app.get("/me")
@guard.auth_required()
async def me(user: Annotated[UserSchema, Depends(get_current_user)]) -> UserSchema:
    return user

@app.get("/articles")
@guard.require_permission(Resource.POST, Permission.READ)
async def list_posts() -> list[PostSchema]:
    return MOCK_DB

@app.post("/articles")
@guard.require_permission(Resource.POST, Permission.WRITE)
async def create_post(data: Annotated[PostCreateSchema, Form]) -> PostSchema:
    pk = sorted(MOCK_DB, key=lambda p: p.id)[-1].id + 1
    post = PostSchema(id=pk, title=data.title)
    MOCK_DB.append(post)
    return post

@app.delete("/articles/{post_id}")
@guard.require_permission(Resource.POST, Permission.DELETE)
async def delete_post(post_id: int) -> dict:
    return {"id": post_id, "deleted": True}

@app.get("/policies")
async def list_policies() -> list[dict]:
    """Inspect current policies from the database."""
    async with async_session() as session:
        result = await session.execute(select(Policy))
        policies = result.scalars().all()
    return [{"sub": p.sub, "obj": p.obj, "act": p.act} for p in policies]

Running

uvicorn main:app --reload

Testing

# View current policies
curl http://localhost:8000/policies

# Login as admin
curl -H "Authorization: Bearer admin" http://localhost:8000/articles

# Delete (only admin has this permission)
curl -X DELETE -H "Authorization: Bearer admin" http://localhost:8000/articles/1

# Viewer cannot delete
curl -X DELETE -H "Authorization: Bearer viewer" http://localhost:8000/articles/1
# → {"detail": "Forbidden"}

If you update rows in the policies table while the app is running, the provider will detect the new policy hash on the next polling cycle and reload the cached enforcer automatically.