airflow operator从URL下载文件并推送到S3？

import requests
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
import boto3

class DownloadAndPushToS3Operator(BaseOperator):
    @apply_defaults
    def __init__(self, url, s3_bucket, s3_key, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.url = url
        self.s3_bucket = s3_bucket
        self.s3_key = s3_key

    def execute(self, context):
        # 下载文件
        response = requests.get(self.url)
        file_content = response.content

        # 推送到S3
        s3 = boto3.client('s3')
        s3.put_object(Body=file_content, Bucket=self.s3_bucket, Key=self.s3_key)

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2022, 1, 1)
}

with DAG('download_and_push_to_s3', default_args=default_args, schedule_interval='@daily') as dag:
    start = DummyOperator(task_id='start')
    download_and_push = DownloadAndPushToS3Operator(
        task_id='download_and_push',
        url='https://example.com/file.txt',
        s3_bucket='my-s3-bucket',
        s3_key='file.txt'
    )
    end = DummyOperator(task_id='end')

    start >> download_and_push >> end