User Context

async def execute(self, input_data, context, user: UserContext):
    if user.is_authenticated:
        return {"message": f"Hello, {user.subject}!"}
    else:
        return {"message": "Please sign in"}

user_id = user.subject  # e.g., "auth0|123456"
user_data = fetch_user_data(user_id)

client = user.client_id
log_access(client_id=client, user_id=user.subject)

user_scopes = user.scopes  # e.g., ["user", "read:data", "write:data"]

if "admin" in user.scopes:
    return admin_dashboard()

email = user.claims.get('email')
name = user.claims.get('name')
custom_data = user.claims.get('custom_field')
role = user.claims.get('https://example.com/role')

if user.has_scope("admin"):
    return {"admin_panel": True}
elif user.has_scope("user"):
    return {"user_panel": True}
else:
    return {"error": "Insufficient permissions"}

if user.has_scope("write:data") and user.has_scope("delete:data"):
    allow_destructive_operations()

from fastapps import BaseWidget, auth_required, UserContext

@auth_required(scopes=["user"])
class UserProfileWidget(BaseWidget):
    identifier = "user-profile"
    title = "User Profile"
    
    async def execute(self, input_data, context, user: UserContext):
        # Check authentication status
        if not user.is_authenticated:
            return {"error": "Authentication required"}
        
        # Access user information
        user_id = user.subject
        email = user.claims.get('email')
        name = user.claims.get('name')
        picture = user.claims.get('picture')
        
        # Check permissions
        is_premium = user.has_scope("premium")
        is_admin = user.has_scope("admin")
        
        # Fetch user data from your database
        user_data = await fetch_user_data(user_id)
        
        return {
            "user_id": user_id,
            "email": email,
            "name": name,
            "picture": picture,
            "is_premium": is_premium,
            "is_admin": is_admin,
            "profile": user_data,
            "scopes": user.scopes
        }

async def execute(self, input_data, context, user: UserContext):
    return {
        "user_id": user.subject,
        "email": user.claims.get('email'),
        "name": user.claims.get('name')
    }

async def execute(self, input_data, context, user: UserContext):
    role = user.claims.get('role', 'user')
    
    if role == 'admin':
        return await self._admin_view(user)
    elif role == 'manager':
        return await self._manager_view(user)
    else:
        return await self._user_view(user)

async def execute(self, input_data, context, user: UserContext):
    # Check multiple permissions
    can_read = user.has_scope("read:data")
    can_write = user.has_scope("write:data")
    can_delete = user.has_scope("delete:data")
    
    return {
        "permissions": {
            "read": can_read,
            "write": can_write,
            "delete": can_delete
        }
    }

async def execute(self, input_data, context, user: UserContext):
    # Access custom claims (Auth0 uses namespaced claims)
    organization = user.claims.get('https://example.com/organization')
    department = user.claims.get('https://example.com/department')
    employee_id = user.claims.get('https://example.com/employee_id')
    
    return {
        "organization": organization,
        "department": department,
        "employee_id": employee_id
    }

from fastapps import optional_auth

@optional_auth(scopes=["user"])
class FlexibleWidget(BaseWidget):
    async def execute(self, input_data, context, user: UserContext):
        # Check if user is authenticated
        if user.is_authenticated:
            # Authenticated version
            return {
                "tier": "premium",
                "user_id": user.subject,
                "email": user.claims.get('email'),
                "features": ["advanced", "export"]
            }
        
        # Anonymous version
        return {
            "tier": "basic",
            "features": ["view"]
        }

@auth_required(scopes=["user"])
class UserDataWidget(BaseWidget):
    async def execute(self, input_data, context, user: UserContext):
        # Always use user.subject for database queries
        user_id = user.subject
        
        # ✅ Good: Use authenticated user ID
        user_data = await fetch_from_database(user_id)
        
        # ❌ Bad: Don't trust user-supplied IDs
        # user_data = await fetch_from_database(input_data.user_id)
        
        return {
            "user_id": user_id,
            "data": user_data
        }

# ✅ Good
async def execute(self, input_data, context, user: UserContext):
    data = fetch_user_data(user.subject)
    return {"data": data}

# ❌ Bad - Don't trust input
async def execute(self, input_data, context, user: UserContext):
    data = fetch_user_data(input_data.user_id)  # UNSAFE!
    return {"data": data}

async def execute(self, input_data, context, user: UserContext):
    # Double-check critical operations
    if not user.has_scope("admin"):
        return {"error": "Unauthorized"}
    
    # Proceed with admin operation
    delete_resource(input_data.resource_id)

async def execute(self, input_data, context, user: UserContext):
    # Log sensitive access
    await log_access(
        user_id=user.subject,
        action="viewed_sensitive_data",
        timestamp=datetime.utcnow()
    )
    
    return {"data": "..."}

async def execute(self, input_data, context, user: UserContext):
    # Always provide defaults for claims
    email = user.claims.get('email', 'unknown')
    name = user.claims.get('name', 'Anonymous')
    
    return {"email": email, "name": name}

from fastapps import UserContext

async def execute(
    self,
    input_data: MyInput,
    context: ClientContext,
    user: UserContext  # Type hint
) -> Dict[str, Any]:
    # IDE will autocomplete user.subject, user.scopes, etc.
    user_id = user.subject
    # ...