气流 - 在循环任务之前添加一个任务，该任务取决于先前的任务

Question

我试图使我的任务依赖关系正确。到目前为止，我有8个任务取决于以前的任务，这很棒，但是我想先在这些任务之前运行一个任务，但是我似乎无法正确正确。在屏幕截图中，您会看到8个任务Initized_costs_to_s3_0 ... 1 ... 2 ........ 7。但是在此之前，我想拥有最新的cost_cost，就像我对交通界的项目一样（请参阅屏幕截图）。这是我针对这两个任务的代码：

我希望它看起来如何：

l = []
for endpoint in ENDPOINTS:
    latest_only = LatestOnlyOperator(
        task_id=f'{endpoint.name}_latest_only',
    )

    if endpoint.name is 'itemized_costs':
        a = []
        # Load each end points data to S3
        for i in range(0, 8):
            s3 = PToS3Operator(
                task_id=f'{endpoint.name}_to_S3_{i}',
                task_no=f'{int(i)}',
                pool_slots=5,
                endpoint=endpoint
            )
            a.append(s3)
            if i not in [0]:
                a[i-1] >> a[i]

    else:
        s3 = PToS3Operator(
            task_id=f'{endpoint.name}_to_S3',
            pool_slots=5,
            endpoint=endpoint
        )
    latest_only >> s3