casbin-fastapi-decorator

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from casbin_fastapi_decorator import AccessSubject, PermissionGuard
from casbin_fastapi_decorator_file import CachedFileEnforcerProvider

# 1. Providers — regular FastAPI dependencies
async def get_current_user() -> dict:
    return {"sub": "alice", "role": "admin"}

# CachedFileEnforcerProvider loads the enforcer once and hot-reloads
# automatically when model.conf or policy.csv changes on disk.
enforcer_provider = CachedFileEnforcerProvider(
    model_path="model.conf",
    policy_path="policy.csv",
)

# 2. Decorator factory
guard = PermissionGuard(
    user_provider=get_current_user,
    enforcer_provider=enforcer_provider,
    error_factory=lambda user, *rv: HTTPException(403, "Forbidden"),
)

# 3. Wire lifespan to start/stop the file watcher
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with enforcer_provider:
        yield

app = FastAPI(lifespan=lifespan)

# 4. Authentication only
@app.get("/me")
@guard.auth_required()
async def me():
    return {"ok": True}

# 5. Static permission check
@app.get("/articles")
@guard.require_permission("articles", "read")
async def list_articles():
    return []

# 6. Dynamic check — object resolved from request
async def get_article(article_id: int) -> dict:
    return {"id": article_id, "owner": "alice"}

@app.get("/articles/{article_id}")
@guard.require_permission(
    AccessSubject(val=get_article, selector=lambda a: a["owner"]),
    "read",
)
async def read_article(article_id: int):
    return {"article_id": article_id}

PermissionGuard(
    user_provider=...,       # FastAPI dependency that returns the current user
    enforcer_provider=...,   # FastAPI dependency that returns a casbin.Enforcer
    error_factory=...,       # callable(user, *rvals) -> Exception
)

AccessSubject(
    val=get_item,                        # FastAPI dependency
    selector=lambda item: item["name"],  # transformation before enforce
)

@app.get("/gallery/vip")
@guard.require_permission(Domain.GALLERY_EMPLOYEE, subject, Action.READ)
@guard.require_permission(Domain.GALLERY_CRIMINAL, subject, Action.READ)
async def route(): ...

from casbin_fastapi_decorator import AccessSubject, AnyOf

@app.get("/gallery")
@guard.require_permission(
    AnyOf(
        (Domain.GALLERY_EMPLOYEE, AccessSubject(get_obj_from_role_permissions), Action.READ),
        (Domain.GALLERY_CRIMINAL, AccessSubject(get_obj_from_role_permissions), Action.READ),
    ),
)
async def route(): ...

def article_not_found_error(user, *resolved_args) -> HTTPException:
    """Return 404 instead of 403 for denied access."""
    return HTTPException(status_code=404, detail="Article not found")

@app.get("/articles/draft")
@guard.require_permission(
    "article", "write",
    error_factory=article_not_found_error,
)
async def read_draft():
    return {"title": "Draft Article"}

from casbin_fastapi_decorator_file import CachedFileEnforcerProvider

enforcer_provider = CachedFileEnforcerProvider(
    model_path="casbin/model.conf",
    policy_path="casbin/policy.csv",
)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    async with enforcer_provider:   # starts watchdog
        yield                       # stops watchdog on shutdown

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

from casbin_fastapi_decorator_db import DatabaseEnforcerProvider

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),
    poll_interval=30.0,  # seconds between DB hash checks
)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    async with enforcer_provider:   # starts watchdog + polling task
        yield

from casbin_fastapi_decorator_casdoor import CasdoorEnforceTarget, CasdoorIntegration

casdoor = CasdoorIntegration(
    endpoint="http://localhost:8000",
    client_id="...", client_secret="...", certificate=cert,
    org_name="my_org", application_name="my_app",
    target=CasdoorEnforceTarget(
        enforce_id=lambda parsed: f"{parsed['owner']}/my_enforcer",
    ),
)
app.include_router(casdoor.router)   # GET /login, GET /callback, POST /logout
guard = casdoor.create_guard()