materialize_psycopg_streaming_subscribe_with_materialized_view.py

1import psycopg
2import os
3
4# Connect to Materialize using environment variables
5# Note: Materialize is wire-compatible with PostgreSQL
6dsn = os.getenv("MZ_DSN", "user=materialize password=password host=localhost port=6875 dbname=materialize")
7
8def main():
9    try:
10        # Establish a connection
11        with psycopg.connect(dsn, autocommit=True) as conn:
12            with conn.cursor() as cur:
13                print("Connected to Materialize!")
14
15                # Create a source-like table for demonstration
16                cur.execute("CREATE TABLE IF NOT EXISTS messages (id INT, content TEXT)")
17                
18                # Insert some sample data
19                cur.execute("INSERT INTO messages VALUES (1, 'Hello'), (2, 'Materialize')")
20                
21                # Create a Materialized View
22                cur.execute("CREATE MATERIALIZED VIEW IF NOT EXISTS msg_count AS SELECT count(*) FROM messages")
23                
24                # Query the view
25                cur.execute("SELECT * FROM msg_count")
26                row = cur.fetchone()
27                print(f"Current count in materialized view: {row[0]}")
28
29                # Demonstrate SUBSCRIBE (Streaming results)
30                print("Starting subscription to 'msg_count'...")
31                with cur.copy("SUBSCRIBE msg_count") as copy:
32                    # In a real streaming app, this would loop indefinitely
33                    # We take the first result to demonstrate the concept
34                    for row in copy:
35                        print(f"Stream Update: {row}")
36                        break
37
38    except Exception as e:
39        print(f"Error: {e}")
40
41if __name__ == "__main__":
42    main()