from builtins import range
from datetime import timedelta

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago

import boto3

args = {
  'owner': 'dpinchukov',
  'start_date': days_ago(0, hour=0, minute=5),
}

dag = DAG(
  dag_id='fetch-from-s3',
  default_args=args,
  schedule_interval=None,
  dagrun_timeout=timedelta(minutes=3),
  tags=['python']
)

def fetch(bucket, key):
  client = boto3.client('s3')
  obj = client.get_object(Bucket=bucket, Key=key)
  print(obj)
  print(obj['Body'].read())

fetch_from_bucket_encrypted_with_aes256 = PythonOperator(
  task_id='fetch-from-bucket-encrypted-with-aes256',
  python_callable=fetch,
  op_kwargs={
    'bucket': 'dev-service1-bucket-encrypted-with-aes256',
    'key': 'emr/data/encrypted.log',
  },
  dag=dag,
)

fetch_from_bucket_encrypted_with_kms = PythonOperator(
  task_id='fetch-from-bucket-encrypted-with-kms',
  python_callable=fetch,
  op_kwargs={
    'bucket': 'dev-service1-bucket-encrypted-with-kms',
    'key': 'emr/data/encrypted.log',
  },
  dag=dag,
)

fetch_from_bucket_encrypted_with_aes256
fetch_from_bucket_encrypted_with_kms