Security Permissions

RESOURCE_ASSET: str
    """
    Asset resource permission constant.
    
    Used for permissions related to asset/dataset management.
    Value: "Assets" or "Datasets" depending on Airflow version
    """

RESOURCE_ASSET_ALIAS: str
    """
    Asset alias resource permission constant.
    
    Used for permissions related to asset alias management.
    Value: "Asset Aliases" or "Dataset Aliases" depending on Airflow version
    """

RESOURCE_BACKFILL: str
    """
    Backfill resource permission constant.
    
    Used for permissions related to backfill operations.
    Value: "Backfills"
    """

RESOURCE_DAG_VERSION: str
    """
    DAG version resource permission constant.
    
    Used for permissions related to DAG version management.
    Value: "DAG Versions"
    """

from airflow.providers.common.compat.security.permissions import (
    RESOURCE_ASSET,
    RESOURCE_ASSET_ALIAS,
    RESOURCE_BACKFILL,
    RESOURCE_DAG_VERSION
)

# Use in security decorators
from airflow.security import permissions
from airflow.www.auth import has_access

# Asset permissions
@has_access(
    permissions=[
        (permissions.ACTION_CAN_READ, RESOURCE_ASSET),
        (permissions.ACTION_CAN_EDIT, RESOURCE_ASSET)
    ]
)
def manage_assets():
    """Function that requires asset read/edit permissions."""
    pass

# Asset alias permissions
@has_access(
    permissions=[
        (permissions.ACTION_CAN_CREATE, RESOURCE_ASSET_ALIAS),
        (permissions.ACTION_CAN_DELETE, RESOURCE_ASSET_ALIAS)
    ]
)
def manage_asset_aliases():
    """Function that requires asset alias create/delete permissions."""
    pass

# Backfill permissions
@has_access(
    permissions=[
        (permissions.ACTION_CAN_CREATE, RESOURCE_BACKFILL),
        (permissions.ACTION_CAN_READ, RESOURCE_BACKFILL)
    ]
)
def create_backfill():
    """Function that requires backfill permissions."""
    pass

# DAG version permissions
@has_access(
    permissions=[
        (permissions.ACTION_CAN_READ, RESOURCE_DAG_VERSION)
    ]
)
def view_dag_versions():
    """Function that requires DAG version read permissions."""
    pass

# Custom permission checking
def check_user_permissions(user, action, resource_type):
    """
    Check if user has permission for specific action on resource.
    
    Args:
        user: User object
        action: Action to check (e.g., permissions.ACTION_CAN_READ)
        resource_type: Resource type (use constants from this module)
    
    Returns:
        bool: True if user has permission, False otherwise
    """
    if resource_type == RESOURCE_ASSET:
        # Check asset permissions
        return user.has_permission(action, RESOURCE_ASSET)
    elif resource_type == RESOURCE_BACKFILL:
        # Check backfill permissions
        return user.has_permission(action, RESOURCE_BACKFILL)
    # ... etc
    
    return False

# Use in role definitions
ASSET_MANAGER_PERMISSIONS = [
    (permissions.ACTION_CAN_READ, RESOURCE_ASSET),
    (permissions.ACTION_CAN_EDIT, RESOURCE_ASSET),
    (permissions.ACTION_CAN_CREATE, RESOURCE_ASSET),
    (permissions.ACTION_CAN_DELETE, RESOURCE_ASSET),
    (permissions.ACTION_CAN_READ, RESOURCE_ASSET_ALIAS),
    (permissions.ACTION_CAN_EDIT, RESOURCE_ASSET_ALIAS),
]

WORKFLOW_MANAGER_PERMISSIONS = [
    (permissions.ACTION_CAN_READ, RESOURCE_BACKFILL),
    (permissions.ACTION_CAN_CREATE, RESOURCE_BACKFILL),
    (permissions.ACTION_CAN_READ, RESOURCE_DAG_VERSION),
]

# Use in Flask-AppBuilder views
from flask_appbuilder import BaseView, expose
from flask_appbuilder.security.decorators import has_access

class AssetView(BaseView):
    @expose('/list/')
    @has_access(
        permissions=[
            (permissions.ACTION_CAN_READ, RESOURCE_ASSET)
        ]
    )
    def list_assets(self):
        """List all assets - requires read permission."""
        return self.render_template('assets/list.html')
    
    @expose('/create/')
    @has_access(
        permissions=[
            (permissions.ACTION_CAN_CREATE, RESOURCE_ASSET)
        ]
    )
    def create_asset(self):
        """Create new asset - requires create permission."""
        return self.render_template('assets/create.html')