📚 课程目录第2章开发环境准备

开发环境准备：虚拟环境搭建、Airflow源码安装、测试框架选择

说实话，很多人在自定义Operator上栽跟头，不是因为写不出代码，而是环境没搭对。我见过太多人直接在全局Python环境里装Airflow，结果依赖冲突搞得一团糟，最后连官方Operator都跑不起来。

这一章，我们就来把开发环境彻底搞定。你跟着我的步骤走，保证后面写代码时顺风顺水。

为什么必须用虚拟环境？

Airflow的依赖管理，说白了就是一场噩梦。它依赖的库版本非常敏感——比如flask、sqlalchemy、werkzeug这些，稍微升级一个次版本就可能崩掉。

我在项目中遇到过最惨的一次：同事直接在服务器上pip install apache-airflow，结果把生产环境的requests库从2.25升级到了2.28，导致另一个服务全部502。嗯，从那以后，我立了个规矩——开发环境必须隔离。

核心原则：每个Airflow项目使用独立的Python虚拟环境，互不干扰。

虚拟环境搭建：两种主流方案

我个人习惯用venv，因为它Python 3自带的，不需要额外装东西。但如果你同时维护多个Airflow版本，conda会更方便。

方案	适用场景	优点	缺点
venv	单版本开发、轻量项目	Python内置、轻量、无额外依赖	Python版本切换不便
conda	多版本Python、复杂依赖	可管理Python版本、跨平台	体积大、启动慢

我建议你选venv，除非你确实需要同时测试Airflow 2.5和2.8。下面我以venv为例，手把手带你搭一遍。

步骤一：创建虚拟环境

# 确保Python版本 >= 3.8（Airflow 2.x 要求）
python3 --version

# 创建虚拟环境（我习惯放在项目根目录下）
python3 -m venv .airflow-venv

# 激活环境
source .airflow-venv/bin/activate   # Linux/Mac
# .airflow-venv\Scripts\activate    # Windows

小技巧：把虚拟环境放在项目目录内，方便多人协作时统一路径。我一般用.airflow-venv这个名字，加个点让它隐藏起来，看着干净。

步骤二：安装Airflow（源码安装）

为什么要源码安装？因为我们要开发自定义Operator，需要能直接修改Airflow的源码来调试。用pip install apache-airflow装的是编译好的包，你改不了源码。

我建议你从GitHub克隆官方仓库，然后以可编辑模式安装。

# 克隆Airflow源码（指定版本，我常用2.8.0）
git clone --branch 2.8.0 https://github.com/apache/airflow.git
cd airflow

# 安装核心依赖（这一步会装很多，耐心等）
pip install -e ".[devel]"

# 验证安装
airflow version
# 输出：2.8.0

你可能会问：为什么装[devel]？因为开发模式包含了测试工具、lint工具、文档生成器等。如果你只是跑跑看，装.[all]也行，但[devel]更全。

注意：源码安装时，pip会尝试编译一些C扩展（比如pandas、numpy）。如果你的系统没有gcc或build-essential，会报错。提前装好：sudo apt install build-essential python3-dev（Linux）或xcode-select --install（Mac）。

步骤三：初始化Airflow数据库

Airflow需要SQLite或PostgreSQL来存储DAG元数据。开发环境用SQLite就够了，不用折腾。

# 设置Airflow主目录（我习惯放在项目里）
export AIRFLOW_HOME=$(pwd)/airflow-home

# 初始化数据库
airflow db init

# 创建管理员用户（方便登录Web UI）
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com \
    --password admin

初始化完成后，你会看到airflow-home目录下生成了airflow.db、airflow.cfg等文件。这些就是Airflow的“家当”。

测试框架选择：为什么是pytest？

写自定义Operator，不写测试等于裸奔。我见过太多人写完Operator直接部署到生产，结果调度器一跑就报错——连最基本的参数校验都没做。

测试框架我选pytest，原因很简单：

简洁：不需要写unittest.TestCase，函数名带test_就行
插件生态丰富：pytest-cov看覆盖率，pytest-xdist并行跑测试
fixture机制：非常适合Airflow这种需要模拟上下文的环境

安装pytest及其常用插件：

pip install pytest pytest-cov pytest-xdist

我的经验：在pyproject.toml或setup.cfg里配置pytest，这样团队所有人都用同一套规则。比如我习惯这样配：

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
addopts = "-v --cov=my_operators --cov-report=term-missing"

这样你只需要跑pytest，就能自动发现测试文件、跑测试、输出覆盖率报告。

测试Airflow Operator的必备工具

光有pytest还不够，你还需要几个Airflow专属的测试工具：

工具/模块	用途	安装方式
`airflow.testing`	提供测试基类、mock上下文	随Airflow源码自带
`pytest-airflow`	pytest插件，简化DAG测试	`pip install pytest-airflow`
`freezegun`	冻结时间，测试调度逻辑	`pip install freezegun`

我个人最常用的是airflow.testing里的DagContext和TaskContext。它们能模拟DAG运行时的上下文，让你在测试里直接调用execute()方法。

环境验证：跑一个最小测试

环境搭好了，我们来验证一下。创建一个测试文件tests/test_env.py：

import pytest
from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator

def test_dag_creation():
    """验证DAG能正常创建"""
    dag = DAG(
        dag_id="test_dag",
        start_date=days_ago(1),
        schedule=None,
    )
    assert dag.dag_id == "test_dag"

def test_python_operator():
    """验证PythonOperator能执行"""
    def my_func():
        return 42

    dag = DAG(dag_id="test_op", start_date=days_ago(1), schedule=None)
    task = PythonOperator(
        task_id="test_task",
        python_callable=my_func,
        dag=dag,
    )
    result = task.execute(context={})
    assert result == 42

然后跑一下：

pytest tests/test_env.py -v

如果看到两个绿色PASSED，恭喜你——环境彻底搞定了。

避坑指南：我曾经在测试时遇到ModuleNotFoundError: No module named 'airflow'。原因很简单——虚拟环境没激活。每次打开新终端，记得先source .airflow-venv/bin/activate。我后来在项目根目录加了个.env文件，用direnv自动激活，省心多了。

本章知识体系总览

下面这张图，帮你把整个开发环境的脉络理清楚：

环境准备就绪，接下来我们就可以正式进入自定义Operator的开发了。记住：环境搭得好，开发没烦恼。别嫌这一步麻烦，后面你会感谢自己的。