Apache Airflow 工作流管理平台

# 使用pyenv管理Python版本pyenv install 3.11.9pyenv global 3.11.9
# 创建虚拟环境python -m venv airflow_venvsource airflow_venv/bin/activate
# 安装Airflowpip install apache-airflow==3.0.0 --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.0.0/constraints-3.11.txt"

# 设置Airflow主目录export AIRFLOW_HOME=~/airflow
# 初始化数据库airflow db init
# 创建管理员用户airflow users create --username admin --firstname Admin --lastname User --role Admin --email admin@example.com

# 启动所有组件(开发模式)airflow standalone
# 访问Web UIhttp://localhost:8080

from datetime import datetimefrom airflow import DAGfrom airflow.operators.bash import BashOperator
with DAG(    dag_id="example_dag",    schedule_interval="@daily",    start_date=datetime(2023, 1, 1),    catchup=False,) as dag:        task1 = BashOperator(        task_id="print_date",        bash_command="date",    )        task2 = BashOperator(        task_id="sleep",        bash_command="sleep 5",    )        task1 >> task2  # 定义任务依赖

import requests
response = requests.post(    "http://localhost:8080/api/v1/dags/example_dag/dagRuns",    json={        "logical_date": "2023-01-01T00:00:00Z",        "conf": {"key": "value"},    },    headers={"Authorization": "Bearer <JWT_TOKEN>"})

class DagRun(models.Model):    """DAG运行实例模型"""        dag_id = models.CharField(max_length=250)    execution_date = models.DateTimeField()    state = models.CharField(max_length=50)    run_id = models.CharField(max_length=250)    external_trigger = models.BooleanField()    conf = models.TextField()  # 存储为JSON        def update_state(self):        """更新DAG运行状态"""        tis = TaskInstance.objects.filter(dag_run=self)        states = [ti.state for ti in tis]                if all(s == State.SUCCESS for s in states):            self.state = State.SUCCESS        elif any(s == State.FAILED for s in states):            self.state = State.FAILED        elif any(s == State.RUNNING for s in states):            self.state = State.RUNNING        else:            self.state = State.QUEUED                    self.save()

class JWTBearer(HTTPBearer):    """JWT认证中间件"""        def __init__(self, **kwargs):        super().__init__(**kwargs)        self.validator = JWTValidator(            required_claims={"aud", "exp", "iat"},
            audience=conf.get_mandatory_list_value("execution_api", "jwt_audience"),        )        async def __call__(self, request: Request):        credentials = await super().__call__(request)        try:            payload = self.validator.validate(credentials.credentials)            request.state.token = TIToken(id=uuid.UUID(payload["sub"]), claims=payload)        except InvalidTokenError as e:            raise HTTPException(status.HTTP_403_FORBIDDEN, str(e))