Scripts¶

"""scripts/my_report.py

Example: Print a greeting to verify the execution environment.
"""
import os

project_root = os.environ.get("DANGO_PROJECT_ROOT", "unknown")
print(f"Hello from Dango! Project root: {project_root}")

import os

import duckdb

project_root = os.environ["DANGO_PROJECT_ROOT"]
db_path = os.path.join(project_root, "data", "warehouse.duckdb")

# Read-only connection (preferred for queries)
conn = duckdb.connect(db_path, read_only=True)
rows = conn.sql(
    "SELECT table_name FROM information_schema.tables"
).fetchall()
conn.close()

"""scripts/import_data.py

Example: Safely write data to the warehouse.
"""
import os
from pathlib import Path

import duckdb

from dango.utils.dbt_lock import DbtLock

project_root = Path(os.environ["DANGO_PROJECT_ROOT"])
db_path = project_root / "data" / "warehouse.duckdb"

lock = DbtLock(project_root)
try:
    lock.acquire(timeout=30)
    conn = duckdb.connect(str(db_path))
    conn.sql("CREATE SCHEMA IF NOT EXISTS script_data")
    conn.sql("""
        CREATE TABLE IF NOT EXISTS script_data.my_import (
            id INTEGER PRIMARY KEY,
            value VARCHAR,
            loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.sql(
        "INSERT INTO script_data.my_import VALUES (1, 'hello', CURRENT_TIMESTAMP)"
    )
    conn.commit()
    conn.close()
    print("Data written successfully.")
finally:
    lock.release()

# BAD: creating tables directly in scripts
conn.sql("CREATE TABLE raw.my_table AS SELECT ...")

# GOOD: define a dbt model in models/
# models/script_models/my_table.sql
# {{ config(materialized='table') }}
# SELECT ...

# BAD: holding the lock during slow network I/O
lock.acquire()
data = fetch_from_api()  # slow!
conn.sql("INSERT INTO ...")

# GOOD: lock only around writes
data = fetch_from_api()
lock.acquire()
conn.sql("INSERT INTO ...")
lock.release()

# BAD: importing internal modules
from dango.config.loader import ConfigLoader  # internal API

# ACCEPTABLE: using the public lock utility
from dango.utils.dbt_lock import DbtLock

"""scripts/email_report.py

Send a daily summary email via SMTP.
"""
import os
import smtplib
from email.message import EmailMessage
from pathlib import Path

import duckdb

project_root = Path(os.environ["DANGO_PROJECT_ROOT"])
db_path = project_root / "data" / "warehouse.duckdb"

smtp_host = os.environ.get("SMTP_HOST", "localhost")
smtp_port = 587
try:
    smtp_port = int(os.environ.get("SMTP_PORT", "587"))
except (ValueError, TypeError):
    pass
smtp_user = os.environ.get("SMTP_USER", "")
smtp_pass = os.environ.get("SMTP_PASS", "")
recipient = os.environ.get("REPORT_RECIPIENT", "[email protected]")

conn = duckdb.connect(str(db_path), read_only=True)
rows = conn.sql("""
    SELECT o.order_date,
           COUNT(*) AS order_count,
           SUM(o.total) AS revenue
    FROM raw.orders o
    WHERE o.order_date >= CURRENT_DATE - INTERVAL '1' DAY
    GROUP BY o.order_date
    ORDER BY o.order_date
""").fetchall()
conn.close()

if not rows:
    print("No orders found for the last 24 hours.")
else:
    body_lines = ["Daily Order Summary", "=" * 40, ""]
    for date, count, revenue in rows:
        body_lines.append(
            f"Date: {date}  Orders: {count}  Revenue: ${revenue:.2f}"
        )

    msg = EmailMessage()
    msg.set_content("\n".join(body_lines))
    msg["Subject"] = f"Daily Report - {rows[0][0]}"
    msg["From"] = smtp_user
    msg["To"] = recipient

    with smtplib.SMTP(smtp_host, smtp_port) as server:
        server.starttls()
        server.login(smtp_user, smtp_pass)
        server.send_message(msg)

    print(f"Report sent to {recipient}")

"""scripts/slack_alert.py

Check data freshness and alert Slack if stale.
"""
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from urllib.request import Request, urlopen

import duckdb

project_root = Path(os.environ["DANGO_PROJECT_ROOT"])
db_path = project_root / "data" / "warehouse.duckdb"
webhook_url = os.environ.get("SLACK_WEBHOOK_URL", "")

if not webhook_url:
    print("SLACK_WEBHOOK_URL not set. Skipping alert.")
    raise SystemExit(0)

conn = duckdb.connect(str(db_path), read_only=True)
# Replace marts.fact_orders with your own model or table
result = conn.sql("""
    SELECT MAX(order_date) AS last_order_date
    FROM marts.fact_orders
    HAVING MAX(order_date) < CURRENT_DATE - INTERVAL '1' DAY
""").fetchall()
conn.close()

if not result or result[0][0] is None:
    print("Data is fresh — no alert needed.")
else:
    last_date = result[0][0]
    days_ago = (datetime.now(timezone.utc).date() - last_date).days
    text = (
        f":warning: *Stale Data Alert*\n"
        f"Latest order date is {last_date} ({days_ago} days ago)."
    )
    payload = json.dumps({"text": text}).encode()
    req = Request(
        webhook_url,
        data=payload,
        headers={"Content-Type": "application/json"},
    )
    urlopen(req)
    print(f"Alert sent: data is {days_ago} days stale.")

"""scripts/sheets_export.py

Export a query result to Google Sheets.
"""
import os
from pathlib import Path

import duckdb
import gspread

project_root = Path(os.environ["DANGO_PROJECT_ROOT"])
db_path = project_root / "data" / "warehouse.duckdb"

conn = duckdb.connect(str(db_path), read_only=True)
df = conn.sql("""
    SELECT order_date,
           COUNT(*) AS orders,
           SUM(amount) AS revenue
    FROM raw.orders
    GROUP BY order_date
    ORDER BY order_date DESC
    LIMIT 100
""").fetchdf()
conn.close()

gc = gspread.service_account()
sh = gc.open(os.environ.get("SHEET_NAME", "Dango Export"))
worksheet = sh.sheet1
worksheet.clear()
worksheet.update([df.columns.values.tolist()] + df.values.tolist())
print(f"Exported {len(df)} rows to sheet '{sh.title}'.")