What switching looks like

from airflow import DAG
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime

def extract():
    hook = PostgresHook("pg_src")
    return hook.get_records(
        "SELECT region, SUM(amount) AS revenue "
        "FROM orders WHERE day = CURRENT_DATE "
        "GROUP BY region")

with DAG("daily_revenue",
         schedule="0 2 * * *",
         start_date=datetime(2026, 1, 1)) as dag:
    PythonOperator(task_id="extract",
                   python_callable=extract)

Pipeline: daily_revenue
Schedule: 0 2 * * *

  [Postgres source: pg_src]
        |
        v
  [SQL transform]
    SELECT region, SUM(amount) AS revenue
    FROM src
    WHERE day = CURRENT_DATE
    GROUP BY region
        |
        v
  [Postgres sink: warehouse.daily_revenue]

# Built visually. No file authoring.

from prefect import flow, task
import pandas as pd
import sqlalchemy as sa

eng = sa.create_engine("postgresql://...")

@task
def transform(df):
    return (df[df.amount > 100]
            .groupby("region")["amount"].sum()
            .reset_index())

@flow(name="revenue_by_region")
def run():
    df = pd.read_sql("SELECT * FROM orders", eng)
    out = transform(df)
    out.to_sql("revenue_gold", eng, if_exists="replace")

Pipeline: revenue_by_region

  [Postgres source]
    SELECT * FROM orders
        |
        v
  [SQL filter]
    SELECT * FROM src WHERE amount > 100
        |
        v
  [SQL aggregate]
    SELECT region, SUM(amount) AS amount
    FROM filter
    GROUP BY region
        |
        v
  [Postgres sink: revenue_gold]

from dagster import asset
import pandas as pd

@asset
def raw_orders():
    return pd.read_sql("SELECT * FROM orders", conn)

@asset
def clean_orders(raw_orders):
    return (raw_orders
            .dropna(subset=["customer_id"])
            .drop_duplicates(subset=["order_id"]))

@asset
def daily_revenue(clean_orders):
    return (clean_orders
            .groupby("day")["amount"].sum()
            .reset_index())

Pipeline: daily_revenue

  [Postgres source: raw_orders]
    SELECT * FROM orders
        |
        v
  [SQL transform: clean_orders]
    SELECT DISTINCT ON (order_id) *
    FROM raw_orders
    WHERE customer_id IS NOT NULL
        |
        v
  [SQL aggregate: daily_revenue]
    SELECT day, SUM(amount) AS amount
    FROM clean_orders
    GROUP BY day

{
  "name": "orders_to_warehouse",
  "nodes": [
    { "name": "HTTP",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "url": "https://api.example.com/orders" }},
    { "name": "Filter",
      "type": "n8n-nodes-base.code",
      "parameters": {
        "jsCode": "return items.filter(
          i => i.json.status === 'paid')" }},
    { "name": "Postgres",
      "type": "n8n-nodes-base.postgres",
      "parameters": { "table": "orders_clean" }}
  ]
}

Pipeline: orders_to_warehouse

  [REST source]
    GET https://api.example.com/orders
        |
        v
  [SQL filter]
    SELECT * FROM src
    WHERE status = 'paid'
        |
        v
  [Postgres sink: orders_clean]