43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
|
import os
|
||
|
|
from datetime import timedelta
|
||
|
|
|
||
|
|
import pendulum
|
||
|
|
from airflow.decorators import dag
|
||
|
|
from airflow.decorators import task
|
||
|
|
|
||
|
|
from common import s3_dowload_unzip_upload
|
||
|
|
|
||
|
|
S3_CONN_ID = os.getenv("S3_CONN_ID", "s3_conn")
|
||
|
|
EXECUTION_TIMEOUT = int(os.getenv("EXECUTION_TIMEOUT", 6))
|
||
|
|
|
||
|
|
default_args = {
|
||
|
|
"execution_timeout": timedelta(hours=EXECUTION_TIMEOUT),
|
||
|
|
"retries": int(os.getenv("DEFAULT_TASK_RETRIES", 1)),
|
||
|
|
"retry_delay": timedelta(seconds=int(os.getenv("DEFAULT_RETRY_DELAY_SECONDS", 60))),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@dag(
|
||
|
|
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
|
||
|
|
schedule=None,
|
||
|
|
catchup=False,
|
||
|
|
default_args=default_args,
|
||
|
|
params={
|
||
|
|
"zipfile": "File to unzip",
|
||
|
|
"src_bucket": "bucket containing the zip file",
|
||
|
|
"dst_bucket": "bucket that will contain unzipped files"
|
||
|
|
},
|
||
|
|
tags=["s3"],
|
||
|
|
)
|
||
|
|
def s3_unzip():
|
||
|
|
@task
|
||
|
|
def unzip(**context):
|
||
|
|
s3_dowload_unzip_upload(S3_CONN_ID,
|
||
|
|
context["params"]["zipfile"],
|
||
|
|
context["params"]["src_bucket"],
|
||
|
|
context["params"]["dst_bucket"])
|
||
|
|
|
||
|
|
unzip()
|
||
|
|
|
||
|
|
|
||
|
|
s3_unzip()
|