New dags

2024-09-20 21:04:35 -04:00
parent a06a53c935
commit 892ab80e3e
4 changed files with 188 additions and 0 deletions
@@ -0,0 +1,47 @@
+
+from airflow import DAG
+from airflow.operators.dummy import DummyOperator
+# from airflow.operators.python import PythonOperator
+from airflow.operators.python_operator import PythonOperator
+from datetime import datetime, timedelta
+from airflow.models import Variable
+
+import requests
+
+default_args = {
+    'owner': 'WrenchBoard',
+    'depends_on_past': False,
+    'start_date': datetime(2022, 1, 1),
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1
+}
+
+
+def WrenchCronAutoMarket01rApi(**kwargs):
+	url = kwargs["api_host"] + "/v1/automarket01"
+	querystring= {"testuser":"ameye"}
+	response = requests.request("GET", url, params=querystring)
+	print(response.text)
+
+with DAG(
+	dag_id="WRENCBOARD_CRON_AUTOMARKET01",
+	default_args=default_args,
+	start_date=datetime(2024, 9, 1, 10, 00),
+	schedule_interval=timedelta(minutes=20),
+	catchup=False) as dag:
+	t1 = PythonOperator(
+		task_id = "wrenchboard_automarket01",
+		python_callable= WrenchCronAutoMarket01rApi,  
+        	op_kwargs={
+            		"api_host" : Variable.get("API_WRENCHBOARD_HOST")
+	        }
+	)
+
+	t2 = DummyOperator(
+		task_id="end"
+	)
+
+	t1 >> t2
+
+
@@ -0,0 +1,47 @@
+
+from airflow import DAG
+from airflow.operators.dummy import DummyOperator
+# from airflow.operators.python import PythonOperator
+from airflow.operators.python_operator import PythonOperator
+from datetime import datetime, timedelta
+from airflow.models import Variable
+
+import requests
+
+default_args = {
+    'owner': 'WrenchBoard',
+    'depends_on_past': False,
+    'start_date': datetime(2022, 1, 1),
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1
+}
+
+
+def WrenchCronProcessTransferApi(**kwargs):
+	url = kwargs["api_host"] + "/v1/processtransfer"
+	querystring= {"testuser":"ameye"}
+	response = requests.request("GET", url, params=querystring)
+	print(response.text)
+
+with DAG(
+	dag_id="WRENCBOARD_CRON_PROCESSTRANSFER",
+	default_args=default_args,
+	start_date=datetime(2024, 9, 1, 10, 00),
+	schedule_interval=timedelta(minutes=60),
+	catchup=False) as dag:
+	t1 = PythonOperator(
+		task_id = "wrenchboard_processtransfer",
+		python_callable= WrenchCronProcessTransferApi,  
+        	op_kwargs={
+            		"api_host" : Variable.get("API_WRENCHBOARD_HOST")
+	        }
+	)
+
+	t2 = DummyOperator(
+		task_id="end"
+	)
+
+	t1 >> t2
+
+
@@ -0,0 +1,47 @@
+
+from airflow import DAG
+from airflow.operators.dummy import DummyOperator
+# from airflow.operators.python import PythonOperator
+from airflow.operators.python_operator import PythonOperator
+from datetime import datetime, timedelta
+from airflow.models import Variable
+
+import requests
+
+default_args = {
+    'owner': 'WrenchBoard',
+    'depends_on_past': False,
+    'start_date': datetime(2022, 1, 1),
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1
+}
+
+
+def WrenchCronScheduleTransferApi(**kwargs):
+	url = kwargs["api_host"] + "/v1/scheduletransfer"
+	querystring= {"testuser":"ameye"}
+	response = requests.request("GET", url, params=querystring)
+	print(response.text)
+
+with DAG(
+	dag_id="WRENCBOARD_CRON_SCHEDULETRANSFER",
+	default_args=default_args,
+	start_date=datetime(2024, 9, 1, 10, 00),
+	schedule_interval=timedelta(minutes=10),
+	catchup=False) as dag:
+	t1 = PythonOperator(
+		task_id = "wrenchboard_scheduletransfer",
+		python_callable= WrenchCronScheduleTransferApi,  
+        	op_kwargs={
+            		"api_host" : Variable.get("API_WRENCHBOARD_HOST")
+	        }
+	)
+
+	t2 = DummyOperator(
+		task_id="end"
+	)
+
+	t1 >> t2
+
+
@@ -0,0 +1,47 @@
+
+from airflow import DAG
+from airflow.operators.dummy import DummyOperator
+# from airflow.operators.python import PythonOperator
+from airflow.operators.python_operator import PythonOperator
+from datetime import datetime, timedelta
+from airflow.models import Variable
+
+import requests
+
+default_args = {
+    'owner': 'WrenchBoard',
+    'depends_on_past': False,
+    'start_date': datetime(2022, 1, 1),
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1
+}
+
+
+def WrenchCronTestNotificationsApi(**kwargs):
+	url = kwargs["api_host"] + "/v1/testnotifications"
+	querystring= {"testuser":"ameye"}
+	response = requests.request("GET", url, params=querystring)
+	print(response.text)
+
+with DAG(
+	dag_id="WRENCBOARD_CRON_TESTNOTIFICATION",
+	default_args=default_args,
+	start_date=datetime(2024, 9, 1, 10, 00),
+	schedule_interval=timedelta(minutes=360),
+	catchup=False) as dag:
+	t1 = PythonOperator(
+		task_id = "wrenchboard_testnotifications",
+		python_callable= WrenchCronTestNotificationsApi,  
+        	op_kwargs={
+            		"api_host" : Variable.get("API_WRENCHBOARD_HOST")
+	        }
+	)
+
+	t2 = DummyOperator(
+		task_id="end"
+	)
+
+	t1 >> t2
+
+