我之前也在单实例Airflow上折腾过多租户的配置，刚好能解答你的疑问，分两部分来说：

一、创建非管理员用户并控制权限 #

Airflow 2.x+ 默认启用RBAC（基于角色的访问控制），自带了几个预设角色，权限从高到低大概是：Admin > User > Op > Viewer > Public。你之前创建的用户都是管理员，大概率是因为没指定--role参数，默认给了Admin角色。

1. 创建非管理员用户（用CLI） #

直接用airflow users create命令，指定非Admin角色就行，比如创建一个普通用户：

airflow users create \
    --username john_doe \
    --firstname John \
    --lastname Doe \
    --email john@example.com \
    --role User \
    --password your_secure_password

预设角色的权限说明：

• Admin：拥有所有权限，能修改系统配置、管理所有用户和DAG
• User：能创建、修改自己的DAG，管理自己的变量/连接（需要额外配置）
• Viewer：只能查看DAG和任务状态，不能修改
• Public：未登录用户的默认角色，一般限制为只读

2. 自定义细粒度权限 #

如果预设角色满足不了需求，你可以创建自定义角色：

• 通过UI操作：登录Admin账户，进入「Security」→「Roles」→「Create」，然后给这个角色分配具体权限（比如只能查看特定DAG、只能读写自己的变量）
• 通过CLI操作：

  # 创建自定义角色
  airflow roles create --role-name custom_user_role
  # 给角色添加权限（比如允许读取DAG）
  airflow roles add-permission --role-name custom_user_role --action can_read --resource dag
  # 给用户分配自定义角色
  airflow users add-role --username john_doe --role-name custom_user_role
  

3. 限制默认注册角色（可选） #

如果允许用户自行注册，要避免默认给Admin权限，修改webserver_config.py里的配置：

# 把默认注册用户的角色设为Viewer
AUTH_ROLE_PUBLIC = "Viewer"
# 开启注册功能（默认可能关闭）
AUTH_REGISTER = True

二、将DAG分配给特定用户并实现过滤 #

Airflow通过DAG的owner参数关联用户，配合权限配置就能实现按所有者过滤。

1. 给DAG指定所有者 #

在定义DAG的时候，明确设置owner参数为目标用户名：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="johns_dag",
    owner="john_doe",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily"
) as dag:
    # 你的任务定义
    pass

2. 开启按所有者过滤的配置 #

修改airflow.cfg里的以下配置：

[webserver]
# 开启按所有者过滤DAG列表
filter_by_owner = True
# 开启基于所有者的访问控制（用户只能访问自己拥有的DAG）
ownership_based_access_control = True

开启后，用户登录后只能看到自己作为owner的DAG，其他用户的DAG会被隐藏。

3. 实现用户专属的变量与连接 #

默认情况下，Variables和Connections是全局的，要让用户只能管理自己的，需要结合RBAC权限：

• 变量：可以给变量命名加前缀（比如john_doe_api_key），然后给用户的角色分配只允许读写带有该前缀的变量的权限；或者在UI中给特定变量设置权限，只允许对应的用户访问。
• 连接：同理，给连接命名加用户前缀，然后通过角色权限限制访问，或者创建自定义角色只允许用户管理自己的连接。

额外：实现用户各自的Python环境 #

单实例Airflow默认共享一个Python环境，要让每个用户的DAG用独立环境，推荐用VirtualenvOperator：

from airflow.operators.python import VirtualenvOperator

def my_task_function():
    # 这里的代码会运行在虚拟环境里
    import pandas as pd
    print(pd.__version__)

virtualenv_task = VirtualenvOperator(
    task_id="run_in_virtualenv",
    python_callable=my_task_function,
    requirements=["pandas==2.2.0"],  # 指定依赖
    system_site_packages=False,  # 不使用全局环境包
    dag=dag
)

这样每个用户的DAG可以指定自己需要的依赖，不会互相影响。如果需要更隔离的环境，也可以用DockerOperator让任务运行在独立容器里。

内容的提问来源于stack exchange，提问作者sebastian