risingwave_psycopg2_streaming_table_materialized_view_quickstart.py

1import psycopg2
2
3# Connect to RisingWave
4# Default parameters: host='localhost', port='4566', user='root', dbname='dev'
5conn = psycopg2.connect(
6    host="localhost",
7    port=4566,
8    user="root",
9    database="dev"
10)
11
12# Open a cursor to perform database operations
13with conn.cursor() as cur:
14    # 1. Create a source table
15    cur.execute("CREATE TABLE IF NOT EXISTS taxi_trips (distance DOUBLE PRECISION, fare DOUBLE PRECISION);")
16
17    # 2. Create a materialized view to perform streaming analysis
18    cur.execute("""
19        CREATE MATERIALIZED VIEW IF NOT EXISTS fare_stats AS
20        SELECT 
21            COUNT(*) as trip_count, 
22            AVG(fare) as avg_fare 
23        FROM taxi_trips;
24    """)
25
26    # 3. Insert some streaming data
27    cur.execute("INSERT INTO taxi_trips (distance, fare) VALUES (2.5, 10.0), (5.0, 20.0);")
28    conn.commit()
29
30    # 4. Query the materialized view
31    cur.execute("SELECT * FROM fare_stats;")
32    result = cur.fetchone()
33    print(f"Trip Count: {result[0]}, Average Fare: {result[1]}")
34
35# Close the connection
36conn.close()