chandu kavar chandulal

Data Engineer at PortOne| Ex-Grab | Ex-ThoughtWorker

chandulal / pull_templated.py

Created November 26, 2019 15:30

	def push_to_xcoms(args, *kwargs):
	value = "dummyValue"
	kwargs['ti'].xcom_push(key="dummyKey", value=value)

	push_to_xcoms_task = PythonOperator(
	task_id='push_to_xcoms',
	provide_context=True,
	python_callable=push_to_xcoms,
	dag=dag
	)

chandulal / push_pull_source.py

Created November 26, 2019 15:27

	def push_to_xcoms(args, *kwargs):
	value = "dummyValue"
	kwargs['ti'].xcom_push(key="dummyKey", value=value)

	def pull_from_xcoms(**kwargs):
	ti = kwargs['ti']
	pulled_value = ti.xcom_pull(key='dummyKey', task_ids='push_to_xcoms')
	print("value=" + str(pulled_value))

	push_to_xcoms_task = PythonOperator(

chandulal / connections.py

Created November 26, 2019 15:06

	self.airflow_api.add_presto_connection("presto-conn",presto_catlog ,presto_schema)

	self.airflow_api.add_mysql_connection("mysql-conn", mysql_database, mysql_user, mysql_password)

chandulal / presto_to_mysql_tests.py

Created September 8, 2019 06:47

	import sys
	import unittest
	import mysql.connector
	import prestodb

	sys.path.append('../')
	from airflow_api import AirflowAPI
	from db_util import DBUtil
	from constants import PRESTO_DB_PORT,MYSQL_DB_PORT

chandulal / presto_to_mysql.py

Created September 8, 2019 06:44

	from airflow.operators.presto_to_mysql import PrestoToMySqlTransfer
	from datetime import datetime
	from airflow import DAG

	default_args = {
	'email': ['[email protected]']
	}

	dag = DAG('presto_to_mysql', description='Presto to Mysql Transfer', default_args=default_args,
	schedule_interval='0 12 * * *',

chandulal / integration_test_hello_world.py

Created September 8, 2019 06:43

	import sys
	import unittest

	sys.path.append('../')
	from airflow_api import AirflowAPI


	class TestHelloWorldDag(unittest.TestCase):
	"""Integration test for Hello world DAG"""

chandulal / xcoms_template_value.py

Created September 8, 2019 06:37

	def test_xcom_in_templated_field(self):
	dag_id = 'hello_world_xcoms'
	dag = self.dagbag.get_dag(dag_id)
	push_to_xcoms_task = dag.get_task('push_to_xcoms')

	execution_date = datetime.now()

	push_to_xcoms_ti = TaskInstance(task=push_to_xcoms_task, execution_date=execution_date)
	context = push_to_xcoms_ti.get_template_context()
	push_to_xcoms_task.execute(context)

chandulal / xcoms_push_pull_test.py

Created September 8, 2019 06:35

	def test_xcoms(self):
	dag_id = 'hello_world_xcoms'
	dag = self.dagbag.get_dag(dag_id)
	push_to_xcoms_task = dag.get_task('push_to_xcoms')
	pull_from_xcoms_task = dag.get_task('pull_from_xcoms')

	execution_date = datetime.now()

	push_to_xcoms_ti = TaskInstance(task=push_to_xcoms_task, execution_date=execution_date)
	context = push_to_xcoms_ti.get_template_context()

chandulal / helloworld_xcoms.py

Created September 8, 2019 06:31

Hello World DAG with Xcoms

	import datetime
	from airflow import DAG
	from airflow.operators import PythonOperator
	from airflow.operators import BashOperator

	yesterday = datetime.datetime.combine(
	datetime.datetime.today() - datetime.timedelta(1),
	datetime.datetime.min.time())

	default_dag_args = {

chandulal / helloworld_sensor_test.py

Created August 9, 2018 11:38

	import unittest
	from datetime import datetime
	from airflow import DAG
	from airflow.models import TaskInstance
	from airflow.operators import HelloworldSensor

	class TestHelloworldSensor(unittest.TestCase):
	def test_poke_should_return_false_when_value_of_minute_is_not_divisible_by_3(self):
	dag = DAG(dag_id='anydag', start_date=datetime.now())
	sensor_task = HelloworldSensor(