I tried to experiment with a helper class that can create a dag with the parameters passed in . but when I try to import the class in my dag file, airflow doesn't pick it up.
Here's my helper class:
from airflow.models.dag import ScheduleInterval
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta, date
class dagClass:
def __init__(self) -> None:
self.dag = None
self.args={
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime.now()-timedelta(days=1),
'email': ['airflow@airflow.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=1),
}
def create_dag(self,dagName):
print(type(self.args))
self.dag=DAG(dagName,
default_args=self.args,
schedule_interval="39 8 * * *")
return self.dag
def get_dag(self, dagName):
dag = self.create_dag(dagName)
print(type(dag))
return dag
def add_task(self,dag,taskID, scriptPath):
return BashOperator(
task_id = taskID,
bash_command = "python3 {}".format(scriptPath),
dag = dag
)
and here's how I am importing it
from dagClass import dagClass
dagObj = dagClass()
dag = dagObj.get_dag("testClass21")
t1 = dagObj.add_task(dag,"t1","test.py")
print(type(dag)) #<class 'airflow.models.dag.DAG'>
print(type(t1)) # <class 'airflow.operators.bash.BashOperator'>
t1
anybody has any idea on what's wrong? airflow is picking up the other dags I created so it isn't a problem with the folder.