Implementing Role-Based Access Control (RBAC) with Auth0 FGA and FastAPI

Most applications start with a need for tenant-level permissions. How do you control what users can do within their own organization? As your application grows, you might find yourself writing complex SQL queries or embedding messy authorization logic directly into your application code just to answer a simple question: "Can this user do this?"

In this article, you’ll learn how to implement a practical, multi-tenant Role-Based Access Control (RBAC) model using Auth0 FGA. We will build a simple API with Python and FastAPI that offloads its authorization decisions to a scalable, centralized service.

What is Role-Based Access Control (RBAC)?

Role-Based Access Control (RBAC) is a method for restricting access to authorized users. Instead of assigning permissions directly to individuals, you grant permissions to roles. This simplifies administration, as you can manage a handful of roles instead of thousands of individual user permissions.

For example, within a tenant organization, you might have:

• An admin role that can manage documents and users.
• A member role that can only read documents.

What is Auth0 FGA

Auth0 FGA is a high-performance and scalable authorization service based on OpenFGA. OpenFGA is an open-source authorization solution inspired by Google's Zanzibar paper that allows you to model, store, and query permissions in a centralized service.

It works with three core concepts:

1. Store: Where you organize your authorization check data
2. Authorization Model: A schema, written in a DSL, that defines the "types" of objects in your system (e.g., user, organization) and the "relations" they can have with each other (e.g., an organization can have a member).
3. Relationship Tuples: The actual data that asserts a relationship. A tuple is a simple statement of fact, such as "user:carla is an admin of organization:acme."

By combining the model and the tuples, Auth0 FGA can answer any access question you ask with very low latency.

Designing a coarse-grained RBAC model

Let's design an authorization model for a multi-tenant application where resources belong to an organization. Users have roles within that organization, and those roles determine their permissions on the resources.

This model is based on the recommended adoption pattern for starting with coarse-grained access control in OpenFGA. This is an approach many companies do when they begin to adopt a more granular system, starting by replicating their existing permissions structure before moving to more granular controls.

Defining types and relations

First, we define our object types: user, organization (our tenant), and resource. Then, we define the relationships between them.

  model   
    schema 1.1

    type user

    type organization
      relations
        define admin: [user]
        define member: [user]

        define can_add_member: admin
        define can_delete_member: admin
        define can_view_member: admin or member
        define can_add_resource: admin or member

    type resource
      relations
        define organization: [organization]

        define can_delete_resource: admin from organization
        define can_view_resource: admin from organization or member from organization

Let's break down this model:

• An organization has two directly assigned roles: admin and member.
• Permissions to manage the organization (like can_add_member or can_view_member) are computed relations based on those roles. For example, only an admin can add a member, but both an admin and a member can view members.
• A resource has a single relation, organization, which links it back to its parent tenant.
• Permissions on a resource (like can_delete_resource) are also computed. They work by looking up the user's role in the linked organization. For a user to have the can_delete_resource permission, they must be an admin of the resource's parent organization. This is what admin from organization means.

The Sample Application

Before we write any code, we need to make sure our environment is ready.

Prerequisites

• Python 3.9+
• An Auth0 FGA account. Sign up for a free account
• An Auth0 account: This will be used to secure your management API.

Get the project

You can clone the project from this GitHub repository.

git clone https://github.com/auth0-blog/auth0-rbac-fga-fastapi

Bear in mind that this project already contains the full working code for this blog post.

Understanding the application

The sample application has the following main entities:

• Organizations: Groups that contain users with specific roles, represented in the model organization.py
• Resources: Assets owned by organizations, represented in the model resource.py
• Users: Individual actors with roles in organizations

• Roles: admin and member, with different permission levels

  For the sake of this code sample, users are represented by strings and stored directly in Auth0 FGA as tuples with their respective relation and objects.

Configure the Auth0 FGA store

The first step is to configure a store in your Auth0 FGA dashboard. A store is a container for your authorization models and relationship data.

1. Log in to your Auth0 FGA dashboard.
2. Click the Create Store button.
3. Give your store a name (e.g., RBAC App) and jurisdiction (e.g. EU), and then click Create.
4. Once your store is created, you'll land on its dashboard. On the left side, click Settings to view your store details. Copy the Store ID and API URL.
5. Scroll down to the Authorized Clients section and click on Create Client. Set a name, select all permissions, and click Create. A modal will open with data containing your Client ID and Client Secret, make sure to copy and store those for later.

Integrating Auth0 FGA into a FastAPI application

Now that your Auth0 FGA store is configured, it's time to set up your FastAPI project and define the authorization rules that govern your application.

Create the authorization model

This is where we define our application's permissions. Create a file named fga/model.fga.yaml and add the authorization model we defined earlier in this blog post. This model defines your object types (user, organization, resource) and the relationships between them. This file also contains tests and tuples to populate your model.

model: |
  model 
    schema 1.1

    type user

    type organization
      relations
        define admin: [user]
        define member: [user]

        define can_add_member: admin
        define can_delete_member: admin
        define can_view_member: admin or member
        define can_add_resource: admin or member

    type resource
      relations
        define organization: [organization]

        define can_delete_resource: admin from organization
        define can_view_resource: admin from organization or member from organization

tuples:
  - user: user:alice
    relation: admin
    object: organization:acme

  - user: user:bob
    relation: member
    object: organization:acme

  - user: user:charlie
    relation: member
    object: organization:acme

  - user: organization:acme
    relation: organization
    object: resource:database-server

  - user: organization:acme
    relation: organization
    object: resource:api-gateway

tests:
  - check:
      - user: user:alice
        object: organization:acme
        assertions:
          can_add_member: true
          can_delete_member: true
          can_view_member: true
          can_add_resource: true

      - user: user:bob
        object: organization:acme
        assertions:
          can_add_member: false
          can_delete_member: false
          can_view_member: true
          can_add_resource: true

      - user: user:alice
        object: resource:database-server
        assertions:
          can_delete_resource: true
          can_view_resource: true

      - user: user:bob
        object: resource:database-server
        assertions:
          can_delete_resource: false
          can_view_resource: true

      - user: user:charlie
        object: resource:api-gateway
        assertions:
          can_delete_resource: false
          can_view_resource: true

You can run tests and interact with this model using the FGA CLI.

Create the Auth0 FGA client

When you created your Auth0 FGA client, you got some credentials. You’ll populate those into a .env file as follows:

AUTH0_FGA_STORE_ID=YOUR_STORE_ID
AUTH0_FGA_API_TOKEN_ISSUER="auth.fga.dev"
AUTH0_FGA_API_AUDIENCE='https://api.eu1.fga.dev/' # changes depending on your region
AUTH0_FGA_API_URL='https://api.eu1.fga.dev' # changes depending on your region
AUTH0_FGA_CLIENT_ID=YOUR_CLIENT_ID
AUTH0_FGA_CLIENT_SECRET=YOUR_CLIENT_SECRET
AUTH0_FGA_AUTHORIZATION_MODEL_ID=YOUR_MODEL_ID

These variables are used in config.py and populated into the app using pydantic-settings.

Next, you will integrate Auth FGA with your FastAPI application. Create a new file app/utils/auth0_fga_client.py and add the following content:

from openfga_sdk import OpenFgaClient
from openfga_sdk.client import ClientConfiguration
from openfga_sdk.client.models import ClientCheckRequest, ClientWriteRequest, ClientTuple
from openfga_sdk.credentials import Credentials, CredentialConfiguration
from app.config import settings
import logging

# Set up logging
logger = logging.getLogger(__name__)

class Auth0FGAClient:
    def __init__(self):
        # Get Auth0 FGA configuration from settings
        if not all([settings.auth0_fga_store_id, settings.auth0_fga_client_id, 
                    settings.auth0_fga_client_secret]):
            raise ValueError("Missing required Auth0 FGA configuration: " \
            "AUTH0_FGA_STORE_ID, AUTH0_FGA_CLIENT_ID, AUTH0_FGA_CLIENT_SECRET")

        # Initialize OpenFgaClient with configuration
        configuration = ClientConfiguration(
            api_url=settings.auth0_fga_api_url,
            store_id=settings.auth0_fga_store_id,
            authorization_model_id=settings.auth0_fga_authorization_model_id,
            credentials= Credentials(
                method="client_credentials",
                configuration= CredentialConfiguration(
                    client_id= settings.auth0_fga_client_id,
                    client_secret= settings.auth0_fga_client_secret,
                    api_audience= settings.auth0_fga_api_audience,
                    api_issuer= settings.auth0_fga_api_token_issuer
                )
            )
        )
        
        self.client = OpenFgaClient(configuration)

    async def check_permission(self, user: str, relation: str, object_id: str) -> bool:
        """Check if a user has a specific relation to an object."""
        try:
            response = await self.client.check(ClientCheckRequest(
                user=user,
                relation=relation,
                object=object_id
            ))
            logger.debug(f"Permission check result: {response.allowed}")
            return response.allowed
        except Exception as e:
            logger.error(f"Error checking permission for user={user}, relation={relation}, object={object_id}: {e}")
            return False

    async def write_tuples(self, tuples: list[ClientTuple]) -> bool:
        """Write relationship tuples to Auth0 FGA."""
        try:
            logger.debug(f"Writing {len(tuples)} tuples to Auth0 FGA")
            write_request = ClientWriteRequest(
                writes=tuples
            )
            
            await self.client.write(write_request)
            logger.debug("Tuples written successfully")
            return True
        except Exception as e:
            logger.error(f"Error writing tuples: {e}")
            return False

    async def delete_tuples(self, tuples: list[ClientTuple]) -> bool:
        """Delete relationship tuples from Auth0 FGA."""
        try:
            logger.debug(f"Deleting {len(tuples)} tuples from Auth0 FGA")
            write_request = ClientWriteRequest(
                deletes=tuples
            )
            await self.client.write(write_request)
            logger.debug("Tuples deleted successfully")
            return True
        except Exception as e:
            logger.error(f"Error deleting tuples: {e}")
            return False
    
    async def list_objects(self, user: str, relation: str, object_type: str) -> list[str]:
        """List all objects of a given type that a user has a specific relation to."""
        try:
            logger.debug(f"Listing objects: user={user}, relation={relation}, type={object_type}")
            response = await self.client.list_objects(
                user=user,
                relation=relation,
                type=object_type
            )
            objects = response.objects if hasattr(response, 'objects') else []
            logger.debug(f"Found {len(objects)} objects")
            return objects
        except Exception as e:
            logger.error(f"Error listing objects for user={user}, relation={relation}, type={object_type}: {e}")
            return []
    
    async def health_check(self) -> bool:
        """Check if the Auth0 FGA service is healthy."""
        try:
            # Simple health check by attempting to read authorization models
            await self.client.read_authorization_models()
            return True
        except Exception as e:
            logger.error(f"Auth0 FGA health check failed: {e}")
            return False

# Global client instance
fga_client = Auth0FGAClient()

Let’s break down what’s happening in the Auth0FGAClient class:

• FGA client initialization: You’re using the openfga-sdk to initialize the client with the .env variables you defined earlier and connect it to the client you created in Auth0 FGA using client-credentials.
• Permission checking: The check_permission() method calls the check request endpoint to verify if a user has a relation with an object.
• Tuple management: The write_tuples() and delete_tuples() methods are used for relationship management using the write tuple method from the SDK.
• Object listing: The list_objects() method uses the list objects endpoint from the SDK to find the resources a user can access.
• Health monitoring: The health_check() method verifies FGA connectivity by requesting to read the authorization models to the SDK.

Create the authorization service

The AuthorizationService class handles the business logic of your application. Create a new file app/services/authorization_service.py and add the following content:

from typing import List, Optional
from openfga_sdk.client.models import ClientTuple
from app.utils.auth0_fga_client import fga_client

ROLES = ["admin", "member"]

class AuthorizationService:
    """RBAC authorization service using Auth0 FGA."""

    async def assign_user_to_organization(self, user_id: str, 
                                          organization_id: str, 
                                          role: str) -> bool:
        """Assign a user to an organization with a specific role (admin or member)."""
        if role not in ROLES:
            raise ValueError(f"Role must be in: {ROLES}")
            
        client_tuple = ClientTuple(
            user=f"user:{user_id}",
            relation=role,
            object=f"organization:{organization_id}"
        )
        return await fga_client.write_tuples([client_tuple])
    
    async def remove_user_from_organization(self, user_id: str, 
                                            organization_id: str, 
                                            role: str) -> bool:
        """Remove a user's role from an organization."""
        if role not in ROLES:
            raise ValueError(f"Role must be in: {ROLES}")
            
        client_tuple = ClientTuple(
            user=f"user:{user_id}",
            relation=role,
            object=f"organization:{organization_id}"
        )
        return await fga_client.delete_tuples([client_tuple])
    
    async def assign_resource_to_organization(self, resource_id: str, 
                                              organization_id: str) -> bool:
        """Assign a resource to an organization."""
        client_tuple = ClientTuple(
            user=f"organization:{organization_id}",
            relation="organization",
            object=f"resource:{resource_id}"
        )
        return await fga_client.write_tuples([client_tuple])
    
    async def check_permission_on_resource(self, user_id: str, action: str, resource_id: str) -> bool:
        """Check if user is allowed to perform an action on a resource."""
        return await fga_client.check_permission(
            user=f"user:{user_id}",
            relation=action,
            object_id=f"resource:{resource_id}"
        )
    
    async def check_permission_on_org(self, user_id: str, action: str, org_id: str) -> bool:
        """Check if user is allowed to perform an action on an organization."""
        return await fga_client.check_permission(
            user=f"user:{user_id}",
            relation=action,
            object_id=f"organization:{org_id}"
        )

    async def get_user_organizations(self, user_id: str) -> List[str]:
        """Get all organizations a user is a member of."""
        admin_orgs = await fga_client.list_objects(
            user=f"user:{user_id}",
            relation="admin",
            object_type="organization"
        )
        member_orgs = await fga_client.list_objects(
            user=f"user:{user_id}",
            relation="member",
            object_type="organization"
        )
        # Remove duplicates and organization: prefix
        all_orgs = list(set(admin_orgs + member_orgs))
        return [org.replace("organization:", "") for org in all_orgs]
    
    async def get_user_resources(self, user_id: str) -> List[str]:
        """Get all resources a user can view."""
        resources = await fga_client.list_objects(
            user=f"user:{user_id}",
            relation="can_view_resource",
            object_type="resource"
        )
        # Remove resource: prefix
        return [res.replace("resource:", "") for res in resources]
    
    async def check_auth0_fga_health(self) -> bool:
        """Check if Auth0 FGA service is healthy."""
        return await fga_client.health_check()

# Global authorization service instance
authz_service = AuthorizationService()

This class interacts with the Auth0FGAClient class and wraps the call into more verbose methods that align with the business logic.

Tuple Management and Access

Any time you create a new organization or a new resource, you need ensure that Auth0 FGA has the correct data. To do this, you write tuples using the methods created in the AuthorizationService class.

Let's take a look at the organization routes and the access control rules you will define:

• GET /organizations - List organizations (member)
• POST /organizations - Create organization (creator becomes admin)
• GET /organizations/{id} - Get organization details (admin or member)
• DELETE /organizations/{id} - Delete organization (admin only)
• POST /organizations/{id}/members - Add member (admin only)
• DELETE /organizations/{id}/members/{user_id} - Remove member (admin only)

To implement these rules, create a new file app/routes/organization_routes.py and add the following content:

from fastapi import APIRouter, HTTPException, Query, Depends
from typing import List
import uuid
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.models.organization import (
    Organization, 
    OrganizationCreate, 
    OrganizationUpdate,
    MemberAssignment
)
from app.database import get_db, OrganizationDB
from app.services.authorization_service import authz_service

router = APIRouter()

@router.get("/", response_model=List[Organization])
async def list_organizations(
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """List all organizations where the user has at least member access."""
    # Get all organizations from database
    result = await db.execute(select(OrganizationDB))
    all_orgs = result.scalars().all()
    
    accessible_orgs = []
    for org in all_orgs:
        if await authz_service.check_permission_on_org(user_id, "can_view_member", org.id):
            accessible_orgs.append(Organization(
                id=org.id,
                name=org.name,
                description=org.description,
                created_at=org.created_at
            ))
    return accessible_orgs

@router.get("/{organization_id}", response_model=Organization)
async def get_organization(
    organization_id: str,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Get a specific organization."""
    if not await authz_service.check_permission_on_org(user_id, 
                                                       "can_view_member", 
                                                       organization_id):
        raise HTTPException(status_code=403, detail="Access denied")
    
    result = await db.execute(select(OrganizationDB).where(OrganizationDB.id == organization_id))
    org_db = result.scalar_one_or_none()
    
    if not org_db:
        raise HTTPException(status_code=404, detail="Organization not found")
    
    return Organization(
        id=org_db.id,
        name=org_db.name,
        description=org_db.description,
        created_at=org_db.created_at
    )

@router.post("/", response_model=Organization)
async def create_organization(
    organization: OrganizationCreate,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Create a new organization. Creator becomes admin."""
    org_id = str(uuid.uuid4())
    
    # Create organization in database
    org_db = OrganizationDB(
        id=org_id,
        name=organization.name,
        description=organization.description,
        created_at=datetime.now()
    )
    
    db.add(org_db)
    await db.commit()
    await db.refresh(org_db)
    
    # Assign creator as admin
    await authz_service.assign_user_to_organization(user_id, org_id, "admin")
    
    return Organization(
        id=org_db.id,
        name=org_db.name,
        description=org_db.description,
        created_at=org_db.created_at
    )

@router.delete("/{organization_id}")
async def delete_organization(
    organization_id: str,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Delete an organization (admin only)."""
    if not await authz_service.check_permission_on_org(user_id, 
                                                       "can_delete_member", 
                                                       organization_id):
        raise HTTPException(status_code=403, detail="Admin access required")
    
    result = await db.execute(select(OrganizationDB).where(OrganizationDB.id == organization_id))
    org_db = result.scalar_one_or_none()
    
    if not org_db:
        raise HTTPException(status_code=404, detail="Organization not found")
    
    await db.delete(org_db)
    await db.commit()
    
    return {"message": "Organization deleted successfully"}

@router.post("/{organization_id}/members")
async def add_member(
    organization_id: str,
    member: MemberAssignment,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Add a member to an organization (admin only)."""
    if not await authz_service.check_permission_on_org(user_id, 
                                                       "can_add_member", 
                                                       organization_id):
        raise HTTPException(status_code=403, detail="Admin access required")
    
    # Check if organization exists in database
    result = await db.execute(select(OrganizationDB).where(OrganizationDB.id == organization_id))
    org_db = result.scalar_one_or_none()
    
    if not org_db:
        raise HTTPException(status_code=404, detail="Organization not found")
    
    if member.role not in ["admin", "member"]:
        raise HTTPException(status_code=400, detail="Role must be 'admin' or 'member'")
    
    success = await authz_service.assign_user_to_organization(
        member.user_id, 
        organization_id, 
        member.role
    )
    
    if not success:
        raise HTTPException(status_code=500, detail="Failed to assign role")
    
    return {"message": f"User {member.user_id} added as {member.role}"}

@router.delete("/{organization_id}/members/{member_user_id}")
async def remove_member(
    organization_id: str,
    member_user_id: str,
    role: str = Query(..., description="Role to remove ('admin' or 'member')"),
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Remove a member from an organization (admin only)."""
    if not await authz_service.check_permission_on_org(user_id, 
                                                       "can_delete_member", 
                                                       organization_id):
        raise HTTPException(status_code=403, detail="Admin access required")
    
    # Check if organization exists in database
    result = await db.execute(select(OrganizationDB).where(OrganizationDB.id == organization_id))
    org_db = result.scalar_one_or_none()
    
    if not org_db:
        raise HTTPException(status_code=404, detail="Organization not found")
    
    if role not in ["admin", "member"]:
        raise HTTPException(status_code=400, detail="Role must be 'admin' or 'member'")
    
    success = await authz_service.remove_user_from_organization(
        member_user_id, 
        organization_id, 
        role
    )
    
    if not success:
        raise HTTPException(status_code=500, detail="Failed to remove role")
    
    return {"message": f"User {member_user_id} removed from {role} role"}

This file uses methods check_permission_on_org and check_permission_on_resource to verify against Auth0 FGA whether the user has the role or permission necessary to perform the action. Otherwise it will return a status of 403 Forbidden.

The endpoints you’re using for creating and deleting an organization use methods assign_user_to_organization and remove_user_from_organization, respectively to write tuples into Auth0 FGA. When adding a new member to an organization, you’re also using the assign_user_to_organization method. Finally to remove a user from an organization, you use the remove_user_from_organization method, which writes to Auth0 FGA and deletes the tuple.

Similarly, to manage resources, you have the following endpoints:

• GET /resources - List accessible resources (admin, member, or organization)
• POST /resources - Create resource (admin or member)
• GET /resources/{id} - Get resource details (admin, member, or organization)
• DELETE /resources/{id} - Delete resource (admin only)
• GET /resources/{id}/permissions - Check user permissions on resource

These endpoints can be implemented in a new file app/routes/resource_routes.py as follows:

from fastapi import APIRouter, HTTPException, Query, Depends
from typing import List
import uuid
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.models.resource import (
    Resource, 
    ResourceCreate, 
    ResourceUpdate
)
from app.database import get_db, ResourceDB
from app.services.authorization_service import authz_service

router = APIRouter()

@router.get("/", response_model=List[Resource])
async def list_resources(
    user_id: str = Query(..., description="User ID for authorization"),
    organization_id: str = Query(None, description="Filter by organization"),
    db: AsyncSession = Depends(get_db)
):
    """List all resources the user can view."""
    # Build query
    query = select(ResourceDB)
    if organization_id:
        query = query.where(ResourceDB.organization_id == organization_id)
    
    result = await db.execute(query)
    all_resources = result.scalars().all()
    
    accessible_resources = []
    for resource in all_resources:
        if await authz_service.check_permission_on_resource(user_id, 
                                                            "can_view_resource", 
                                                            resource.id):
            accessible_resources.append(Resource(
                id=resource.id,
                name=resource.name,
                description=resource.description,
                resource_type=resource.resource_type,
                organization_id=resource.organization_id,
                created_at=resource.created_at
            ))
    
    return accessible_resources

@router.get("/{resource_id}", response_model=Resource)
async def get_resource(
    resource_id: str,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Get a specific resource."""
    if not await authz_service.check_permission_on_resource(user_id, 
                                                            "can_view_resource", 
                                                            resource_id):
        raise HTTPException(status_code=403, detail="Access denied")
    
    result = await db.execute(select(ResourceDB).where(ResourceDB.id == resource_id))
    resource_db = result.scalar_one_or_none()
    
    if not resource_db:
        raise HTTPException(status_code=404, detail="Resource not found")
    
    return Resource(
        id=resource_db.id,
        name=resource_db.name,
        description=resource_db.description,
        resource_type=resource_db.resource_type,
        organization_id=resource_db.organization_id,
        created_at=resource_db.created_at
    )

@router.post("/", response_model=Resource)
async def create_resource(
    resource: ResourceCreate,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Create a new resource (admin or member of organization)."""
    if not await authz_service.check_permission_on_org(user_id, 
                                                       "can_add_resource", 
                                                       resource.organization_id):
        raise HTTPException(status_code=403, detail="Cannot add resources to this organization")
    
    resource_id = str(uuid.uuid4())
    
    # Create resource in database
    resource_db = ResourceDB(
        id=resource_id,
        name=resource.name,
        description=resource.description,
        resource_type=resource.resource_type,
        organization_id=resource.organization_id,
        created_at=datetime.now()
    )
    
    db.add(resource_db)
    await db.commit()
    await db.refresh(resource_db)
    
    # Link resource to organization in OpenFGA
    await authz_service.assign_resource_to_organization(resource_id, resource.organization_id)
    
    return Resource(
        id=resource_db.id,
        name=resource_db.name,
        description=resource_db.description,
        resource_type=resource_db.resource_type,
        organization_id=resource_db.organization_id,
        created_at=resource_db.created_at
    )

@router.delete("/{resource_id}")
async def delete_resource(
    resource_id: str,
    user_id: str = Query(..., description="User ID for authorization"),
    db: AsyncSession = Depends(get_db)
):
    """Delete a resource (admin only)."""
    if not await authz_service.check_permission_on_resource(user_id, 
                                                            "can_delete_resource", 
                                                            resource_id):
        raise HTTPException(status_code=403, detail="Admin access required")
    
    result = await db.execute(select(ResourceDB).where(ResourceDB.id == resource_id))
    resource_db = result.scalar_one_or_none()
    
    if not resource_db:
        raise HTTPException(status_code=404, detail="Resource not found")
    
    await db.delete(resource_db)
    await db.commit()
    
    return {"message": "Resource deleted successfully"}

@router.get("/{resource_id}/permissions")
async def check_resource_permissions(
    resource_id: str,
    user_id: str = Query(..., description="User ID to check permissions for"),
    db: AsyncSession = Depends(get_db)
):
    """Check what permissions a user has on a specific resource."""
    result = await db.execute(select(ResourceDB).where(ResourceDB.id == resource_id))
    resource_db = result.scalar_one_or_none()
    
    if not resource_db:
        raise HTTPException(status_code=404, detail="Resource not found")
    
    permissions = {
        "can_view": await authz_service.check_permission_on_resource(user_id, "can_view_resource", resource_id),
        "can_delete": await authz_service.check_permission_on_resource(user_id, "can_delete_resource", resource_id)
    }
    
    return {
        "resource_id": resource_id,
        "user_id": user_id,
        "permissions": permissions
    }

Similarly to the organization routes class, the resources routes use the check_permission_on_resource method from the AuthorizationService class to verify whether the user has the necessary role or permission to perform the action. Otherwise, the method returns a status of 403 Forbidden.

If you want to test the endpoints from the code sample, you can use this Postman collection or make requests using cURL. If you run the code sample on your localhost, you'll find the documentation for the API is available at: http://localhost:8000/docs

Conclusion

In this blog post, we showed how to implement RBAC in a FastAPI application, and integrated it with Auth0 FGA using the Python SDK.

You have now modeled a common multi-tenant RBAC pattern, externalizing your authorization logic from your application code. This approach offers significant advantages, including improved scalability, clarity, and maintainability. By separating your authorization rules from your business logic, your API endpoints become cleaner, and you can update your permission model without rewriting application code.

The coarse-grained model you built is the perfect foundation for adding more complex, fine-grained permissions as your application's needs evolve.

Check out the documentation to learn more about Auth0 FGA.