如何用AirflowSecurityManager在Apache Airflow中创建带指定权限的自定义角色

核心实现思路 #

利用Airflow内置的SecurityManager和Flask-AppBuilder的ORM模型，直接操作元数据数据库来创建角色并关联权限。以下是针对你提供的roles_permissions字典的完整实现代码：

完整代码示例 #

from airflow.security import SecurityManager
from airflow.www.app import create_app
from flask_appbuilder.security.sqla.models import Role, PermissionView
from sqlalchemy.exc import IntegrityError

# 初始化Airflow应用与安全管理组件
app = create_app()
security_manager = SecurityManager(app)
db_session = security_manager.get_session()

# 你的角色-权限映射字典
roles_permissions = {
    "Role1": ["can_dag_read", "can_dag_edit"],
    "Role2": ["can_dag_read"]
}

# 遍历创建角色并关联权限
for role_name, perm_names in roles_permissions.items():
    # 检查角色是否已存在
    existing_role = db_session.query(Role).filter(Role.name == role_name).first()
    if existing_role:
        print(f"角色 {role_name} 已存在，跳过创建")
        continue
    
    # 创建新角色
    new_role = Role(name=role_name)
    db_session.add(new_role)
    
    # 关联指定权限
    for perm_name in perm_names:
        # 查找对应权限的PermissionView实例
        perm_view = db_session.query(PermissionView).filter(PermissionView.permission_name == perm_name).first()
        if not perm_view:
            print(f"权限 {perm_name} 不存在，跳过关联")
            continue
        new_role.permissions.append(perm_view)
    
    # 提交事务并处理异常
    try:
        db_session.commit()
        print(f"成功创建角色 {role_name} 并完成权限关联")
    except IntegrityError:
        db_session.rollback()
        print(f"创建角色 {role_name} 失败：角色已存在或数据库冲突")

关键细节说明 #

• Airflow上下文依赖：必须通过create_app()初始化Airflow的Flask应用，否则SecurityManager无法正常获取数据库会话和权限模型
• 权限匹配逻辑：Airflow的权限以PermissionView实例存在，通过permission_name字段匹配你提供的字符串（如can_dag_read）
• 异常处理：添加IntegrityError捕获，避免因角色重复创建导致程序崩溃
• 已有角色更新：如果需要更新现有角色的权限，可修改逻辑：若角色存在，先清空existing_role.permissions再添加新权限，或按需合并权限

运行注意事项 #

• 需在Airflow运行环境中执行脚本（可通过airflow python your_script.py命令运行）
• 确保执行脚本的用户拥有Airflow元数据数据库的读写权限
• 可通过Airflow UI的「Security > Permissions」页面确认所有可用的权限名称

内容的提问来源于stack exchange，提问作者Náthali