Pythonsql_patterns.py
Copy
1
import sqlite3
2
 
3
conn = sqlite3.connect(":memory:")
4
conn.row_factory = sqlite3.Row
5
cur = conn.cursor()
6
 
7
# Setup
8
cur.executescript("""
9
CREATE TABLE events (
10
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
11
    user_id    INTEGER NOT NULL,
12
    event_type TEXT NOT NULL,
13
    created_at TEXT NOT NULL DEFAULT (datetime('now'))
14
);
15
 
16
CREATE TABLE daily_revenue (
17
    date    TEXT PRIMARY KEY,
18
    revenue REAL NOT NULL
19
);
20
 
21
INSERT INTO daily_revenue (date, revenue) VALUES
22
    ('2024-01-01', 1200), ('2024-01-02', 980), ('2024-01-03', 1500),
23
    ('2024-01-04', 750), ('2024-01-05', 1100), ('2024-01-06', 1350),
24
    ('2024-01-07', 890), ('2024-01-08', 1600);
25
""")
26
 
27
# Pattern 1: UPSERT
28
cur.executemany("INSERT INTO events (user_id, event_type, created_at) VALUES (?, ?, ?)", [
29
    (1, "login", "2024-01-10 08:00:00"),
30
    (1, "login", "2024-01-10 09:00:00"),  # duplicate user
31
    (2, "signup", "2024-01-10 08:30:00"),
32
])
33
conn.commit()
34
 
35
# Pattern 2: Deduplicate — keep only latest per user
36
cur.execute("""
37
SELECT e.* FROM events e
38
INNER JOIN (
39
    SELECT user_id, MAX(created_at) AS latest
40
    FROM events
41
    GROUP BY user_id
42
) latest ON e.user_id = latest.user_id AND e.created_at = latest.latest;
43
""")
44
print("Latest event per user:")
45
for row in cur.fetchall():
46
    print(f"  user {row['user_id']}: {row['event_type']} at {row['created_at']}")
47
 
48
# Pattern 3: 7-day rolling average
49
cur.execute("""
50
SELECT date, revenue,
51
       ROUND(AVG(revenue) OVER (
52
           ORDER BY date
53
           ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
54
       ), 2) AS rolling_7d_avg
55
FROM daily_revenue
56
ORDER BY date;
57
""")
58
print("\n7-day rolling revenue average:")
59
for row in cur.fetchall():
60
    print(f"  {row[0]}: ${row[1]} | 7d avg: ${row[2]}")
61
 
62
# Pattern 4: Running total
63
cur.execute("""
64
SELECT date,
65
       revenue,
66
       ROUND(SUM(revenue) OVER (ORDER BY date ROWS UNBOUNDED PRECEDING), 2) AS cumulative_total
67
FROM daily_revenue
68
ORDER BY date;
69
""")
70
print("\nCumulative revenue:")
71
for row in cur.fetchall():
72
    print(f"  {row[0]}: ${row[1]} | cumulative: ${row[2]}")
73
 
74
conn.close()
75