airflow_dag_sql_execute_query_operator_create_insert.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
5
6with DAG(
7    dag_id="example_common_sql",
8    start_date=datetime.datetime(2023, 1, 1),
9    schedule="@daily",
10    catchup=False,
11) as dag:
12
13    # Example of executing a simple query using the SQLExecuteQueryOperator
14    create_table = SQLExecuteQueryOperator(
15        task_id="create_table",
16        conn_id="sql_default",
17        sql="""
18            CREATE TABLE IF NOT EXISTS users (
19                id SERIAL PRIMARY KEY,
20                name VARCHAR(50),
21                created_at TIMESTAMP
22            );
23        """,
24    )
25
26    insert_data = SQLExecuteQueryOperator(
27        task_id="insert_data",
28        conn_id="sql_default",
29        sql="""
30            INSERT INTO users (name, created_at)
31            VALUES ('airflow_user', '{{ ts }}');
32        """,
33    )
34
35    create_table >> insert_data