# Python approach — 6 steps, 40 lines
import duckdb

con = duckdb.connect("pipeline.duckdb")

# Step 1: Load raw data
con.execute("CREATE TABLE raw AS SELECT * FROM read_csv('data.csv')")

# Step 2: Clean data
con.execute("""
 CREATE TABLE cleaned AS
 SELECT * FROM raw
 WHERE amount > 0 AND name IS NOT NULL
""")

# Step 3: Transform
con.execute("""
 CREATE TABLE transformed AS
 SELECT category, SUM(amount) as total, COUNT(*) as cnt
 FROM cleaned
 GROUP BY category
""")

# Step 4: Insert into target
con.execute("INSERT INTO summary SELECT * FROM transformed")

# Step 5: Export report
con.execute("COPY (SELECT * FROM transformed) TO 'report.csv'")

# Step 6: Clean up temp tables
con.execute("DROP TABLE raw")
con.execute("DROP TABLE cleaned")
con.execute("DROP TABLE transformed")

WITH
 raw AS (SELECT * FROM read_csv('data.csv')),
 cleaned AS (
 SELECT * FROM raw
 WHERE amount > 0 AND name IS NOT NULL
 ),
 transformed AS (
 SELECT category, SUM(amount) as total, COUNT(*) as cnt
 FROM cleaned
 GROUP BY category
 ),
 -- Now the magic: DML inside CTEs!
 _insert AS INSERT INTO summary SELECT * FROM transformed,
 _export AS COPY (SELECT * FROM transformed) TO 'report.csv'
SELECT * FROM transformed;

WITH
 step1 AS (INSERT INTO table_a SELECT * FROM source),
 step2 AS (UPDATE table_b SET status = 'done' WHERE id IN (SELECT id FROM table_a)),
 step3 AS (DELETE FROM staging WHERE created_at < '2026-01-01'),
 step4 AS (COPY (SELECT * FROM table_b) TO 'output.csv')
SELECT count(*) FROM table_b;

WITH
 -- Load raw sales data
 raw_sales AS (
 SELECT * FROM read_csv_auto('sales_2026_08.csv')
 ),

 -- Clean: filter invalid records
 cleaned AS (
 SELECT * FROM raw_sales
 WHERE amount > 0
 AND product IS NOT NULL
 AND order_date >= '2026-08-01'
 ),

 -- Transform: aggregate by category
 aggregated AS (
 SELECT
 category,
 SUM(amount) AS total_revenue,
 COUNT(*) AS order_count,
 AVG(amount) AS avg_order_value
 FROM cleaned
 GROUP BY category
 ),

 -- Insert into the production table
 _upsert AS INSERT INTO daily_sales_summary
 SELECT * FROM aggregated
 ON CONFLICT (category) DO UPDATE SET
 total_revenue = EXCLUDED.total_revenue,
 order_count = EXCLUDED.order_count,
 avg_order_value = EXCLUDED.avg_order_value,

 -- Export a report for the business team
 _report AS COPY (
 SELECT category, total_revenue, order_count
 FROM aggregated
 ORDER BY total_revenue DESC
 ) TO 'daily_sales_report.csv'

-- Final result: see today's summary
SELECT * FROM aggregated ORDER BY total_revenue DESC;