Alibaba Cloud OSS Operators¶

Overview¶

Airflow to Alibaba Cloud Object Storage Service (OSS) integration provides several operators to create and interact with OSS buckets.

OSSKeySensor

OSSCreateBucketOperator

OSSDeleteBucketOperator

OSSUploadObjectOperator

OSSDownloadObjectOperator

OSSDeleteBatchObjectOperator

OSSDeleteObjectOperator

Create and Delete Alibaba Cloud OSS Buckets¶

Purpose¶

This example dag uses OSSCreateBucketOperator and OSSDeleteBucketOperator to create a new OSS bucket with a given bucket name then delete it.

Defining tasks¶

In the following code we create a new bucket and then delete the bucket.

tests/system/alibaba/example_oss_bucket.py[source]

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2021, 1, 1),
    schedule=None,
    default_args={"bucket_name": "your bucket", "region": "your region"},
    max_active_runs=1,
    tags=["example"],
    catchup=False,
) as dag:
    create_bucket = OSSCreateBucketOperator(task_id="task1", region=REGION)

    delete_bucket = OSSDeleteBucketOperator(task_id="task2", region=REGION)

    create_bucket >> delete_bucket

    from tests_common.test_utils.watcher import watcher

    # This test needs watcher in order to properly mark success/failure
    # when "tearDown" task with trigger rule is part of the DAG
    list(dag.tasks) >> watcher()