examples

Author: millerjp

Makefile

@@ -287,6 +287,50 @@ We welcome contributions! Please see:
 - [FastAPI Integration](examples/fastapi_app/README.md) - Complete REST API example
 - [More Examples](examples/) - Additional usage patterns
 
+## 🎯 Running the Examples
+
+The project includes comprehensive examples demonstrating various features and use cases. Each example can be run using the provided Makefile, which automatically handles Cassandra setup if needed.
+
+### Available Examples
+
+Run any example with: `make example-<name>`
+
+- **`make example-basic`** - Basic connection and query execution
+- **`make example-streaming`** - Memory-efficient streaming of large result sets with True Async Paging
+- **`make example-context-safety`** - Demonstrates proper context manager usage and resource isolation
+- **`make example-export-large-table`** - Export large tables to CSV with progress tracking
+- **`make example-export-parquet`** - Export data to Parquet format with complex data types
+- **`make example-metrics`** - Comprehensive metrics collection and performance monitoring
+- **`make example-metrics-simple`** - Basic metrics collection example
+- **`make example-realtime`** - Real-time data processing with sliding window analytics
+- **`make example-streaming-demo`** - Visual demonstration that streaming doesn't block the event loop
+
+### Running with External Cassandra
+
+If you have Cassandra running elsewhere:
+
+```bash
+# Single node
+CASSANDRA_CONTACT_POINTS=10.0.0.1 make example-streaming
+
+# Multiple nodes
+CASSANDRA_CONTACT_POINTS=10.0.0.1,10.0.0.2,10.0.0.3 make example-streaming
+
+# With custom port
+CASSANDRA_CONTACT_POINTS=cassandra.example.com CASSANDRA_PORT=9043 make example-basic
+```
+
+### Example Descriptions
+
+- **Basic Example**: Shows fundamental operations like connecting, executing queries, and using prepared statements
+- **Streaming Examples**: Demonstrate True Async Paging for processing millions of rows without memory issues
+- **Export Examples**: Show how to export Cassandra data to various formats (CSV, Parquet) with progress tracking
+- **Metrics Examples**: Illustrate performance monitoring, query tracking, and connection health checking
+- **Real-time Processing**: Demonstrates processing time-series IoT data with concurrent operations
+- **Context Safety Demo**: Proves that errors in one operation don't affect others when using context managers
+
+Each example includes detailed comments explaining the concepts and best practices. Start with `example-basic` if you're new to the library.
+
 ## ⚡ Performance
 
 async-cassandra enables your async Python application to work with Cassandra without blocking the event loop. While it doesn't eliminate the underlying driver's thread pool, it prevents those blocking operations from freezing your entire application. This is crucial for web servers where a blocked event loop means no requests can be processed.

README.md

@@ -40,26 +40,33 @@
 
 async def demonstrate_query_error_safety(cluster):
     """Show that query errors don't close the session."""
-    logger.info("\n=== Demonstrating Query Error Safety ===")
+    logger.info("\n" + "=" * 80)
+    logger.info("🛡️  QUERY ERROR SAFETY DEMONSTRATION")
+    logger.info("=" * 80)
 
     async with await cluster.connect() as session:
+        logger.info("\n🧪 Test 1: Execute a failing query")
         try:
             # This will fail
             await session.execute("SELECT * FROM non_existent_table")
         except InvalidRequest as e:
-            logger.info(f"Query failed as expected: {e}")
+            logger.info(f"   ✓ Query failed as expected: {type(e).__name__}")
 
         # Session should still work
-        logger.info("Session still works after error:")
+        logger.info("\n🧪 Test 2: Verify session still works after error")
         result = await session.execute("SELECT release_version FROM system.local")
-        logger.info(f"Cassandra version: {result.one().release_version}")
+        logger.info(f"   ✅ Session is healthy! Cassandra version: {result.one().release_version}")
+        logger.info("\n💡 Key insight: Query errors are isolated - they don't affect the session!")
 
 
 async def demonstrate_streaming_error_safety(cluster):
     """Show that streaming errors don't close the session."""
-    logger.info("\n=== Demonstrating Streaming Error Safety ===")
+    logger.info("\n" + "=" * 80)
+    logger.info("🌊 STREAMING ERROR SAFETY DEMONSTRATION")
+    logger.info("=" * 80)
 
     async with await cluster.connect() as session:
+        logger.info("\n🛠️  Setting up test data...")
         # Create test keyspace and data
         await session.execute(
             """
@@ -70,96 +77,108 @@ async def demonstrate_streaming_error_safety(cluster):
             }
             """
         )
-        await session.set_keyspace("context_demo")
-
         await session.execute(
             """
-            CREATE TABLE IF NOT EXISTS test_data (
+            CREATE TABLE IF NOT EXISTS context_demo.test_data (
                 id UUID PRIMARY KEY,
                 value TEXT
             )
             """
         )
 
         # Insert some data using prepared statement
-        insert_stmt = await session.prepare("INSERT INTO test_data (id, value) VALUES (?, ?)")
+        insert_stmt = await session.prepare(
+            "INSERT INTO context_demo.test_data (id, value) VALUES (?, ?)"
+        )
         for i in range(10):
             await session.execute(insert_stmt, [uuid.uuid4(), f"value_{i}"])
+        logger.info("   ✓ Created 10 test records")
 
         # Try streaming from non-existent table (will fail)
+        logger.info("\n🧪 Test 1: Stream from non-existent table")
         try:
             async with await session.execute_stream("SELECT * FROM non_existent_table") as stream:
                 async for row in stream:
                     pass
         except Exception as e:
-            logger.info(f"Streaming failed as expected: {e}")
+            logger.info(f"   ✓ Streaming failed as expected: {type(e).__name__}")
 
         # Session should still work for new streaming
-        logger.info("Starting new streaming operation after error:")
+        logger.info("\n🧪 Test 2: Start new streaming operation after error")
         count = 0
-        async with await session.execute_stream("SELECT * FROM test_data") as stream:
+        async with await session.execute_stream("SELECT * FROM context_demo.test_data") as stream:
             async for row in stream:
                 count += 1
 
-        logger.info(f"Successfully streamed {count} rows after error")
+        logger.info(f"   ✅ Successfully streamed {count} rows after error!")
+        logger.info("\n💡 Key insight: Streaming errors are isolated - session remains healthy!")
 
         # Cleanup
         await session.execute("DROP KEYSPACE context_demo")
 
 
 async def demonstrate_context_manager_isolation(cluster):
     """Show how context managers isolate resource cleanup."""
-    logger.info("\n=== Demonstrating Context Manager Isolation ===")
+    logger.info("\n" + "=" * 80)
+    logger.info("🔒 CONTEXT MANAGER ISOLATION DEMONSTRATION")
+    logger.info("=" * 80)
 
     # Scenario 1: Session context doesn't affect cluster
-    logger.info("\nScenario 1: Session context with error")
+    logger.info("\n🧪 Scenario 1: Session error doesn't affect cluster")
     try:
         async with await cluster.connect() as session:
             result = await session.execute("SELECT now() FROM system.local")
-            logger.info(f"Query succeeded: {result.one()[0]}")
+            logger.info(f"   ✓ Query succeeded: {result.one()[0]}")
+            logger.info("   💥 Simulating error...")
             raise ValueError("Simulated error in session context")
     except ValueError:
-        logger.info("Error handled, session was closed by context manager")
+        logger.info("   ✓ Error handled, session closed by context manager")
 
     # Cluster should still work
-    logger.info("Creating new session from same cluster:")
+    logger.info("\n🧪 Creating new session from same cluster:")
     async with await cluster.connect() as session2:
         result = await session2.execute("SELECT now() FROM system.local")
-        logger.info(f"New session works: {result.one()[0]}")
+        logger.info(f"   ✅ New session works perfectly: {result.one()[0]}")
 
     # Scenario 2: Streaming context doesn't affect session
-    logger.info("\nScenario 2: Streaming context with early exit")
+    logger.info("\n🧪 Scenario 2: Early streaming exit doesn't affect session")
     async with await cluster.connect() as session3:
         # Stream with early exit
         count = 0
+        logger.info("   🔄 Starting streaming with early exit...")
         async with await session3.execute_stream("SELECT * FROM system.local") as stream:
             async for row in stream:
                 count += 1
+                logger.info(f"   ✓ Read {count} row, exiting early...")
                 break  # Early exit
 
-        logger.info(f"Exited streaming early after {count} row")
-
         # Session should still work
+        logger.info("\n   🧪 Testing session after early streaming exit:")
         result = await session3.execute("SELECT now() FROM system.local")
-        logger.info(f"Session still works: {result.one()[0]}")
+        logger.info(f"   ✅ Session still healthy: {result.one()[0]}")
+
+    logger.info("\n💡 Key insight: Context managers provide proper isolation!")
 
 
 async def demonstrate_concurrent_safety(cluster):
     """Show that multiple operations can use shared resources safely."""
-    logger.info("\n=== Demonstrating Concurrent Safety ===")
+    logger.info("\n" + "=" * 80)
+    logger.info("🚀 CONCURRENT OPERATIONS SAFETY DEMONSTRATION")
+    logger.info("=" * 80)
 
     # Create shared session
+    logger.info("\n🔄 Running multiple concurrent operations on shared session...")
     async with await cluster.connect() as session:
 
         async def worker(worker_id, query_count):
             """Worker that executes queries."""
             for i in range(query_count):
                 try:
-                    result = await session.execute("SELECT now() FROM system.local")
-                    logger.info(f"Worker {worker_id} query {i+1}: {result.one()[0]}")
+                    await session.execute("SELECT now() FROM system.local")
+                    logger.info(f"   👷 Worker {worker_id} query {i+1}: Success")
                     await asyncio.sleep(0.1)
                 except Exception as e:
-                    logger.error(f"Worker {worker_id} error: {e}")
+                    logger.error(f"   ❌ Worker {worker_id} error: {e}")
 
         async def streamer():
             """Worker that uses streaming."""
@@ -171,27 +190,30 @@ async def streamer():
                     async for row in stream:
                         count += 1
                         if count % 5 == 0:
-                            logger.info(f"Streamer: Processed {count} keyspaces")
+                            logger.info(f"   🌊 Streamer: Processed {count} keyspaces")
                             await asyncio.sleep(0.1)
-                    logger.info(f"Streamer: Total {count} keyspaces")
+                    logger.info(f"   ✅ Streamer: Completed ({count} keyspaces)")
             except Exception as e:
-                logger.error(f"Streamer error: {e}")
+                logger.error(f"   ❌ Streamer error: {e}")
 
         # Run workers concurrently
         await asyncio.gather(worker(1, 3), worker(2, 3), streamer(), return_exceptions=True)
 
-        logger.info("All concurrent operations completed")
+        logger.info("\n✅ All concurrent operations completed successfully!")
+        logger.info("\n💡 Key insight: Multiple operations can safely share a session!")
 
 
 async def main():
     """Run all demonstrations."""
-    logger.info("Starting Context Manager Safety Demonstration")
+    logger.info("\n" + "=" * 80)
+    logger.info("🛡️  CONTEXT MANAGER SAFETY DEMONSTRATION")
+    logger.info("=" * 80)
 
     # Get contact points from environment or use localhost
     contact_points = os.environ.get("CASSANDRA_CONTACT_POINTS", "localhost").split(",")
     port = int(os.environ.get("CASSANDRA_PORT", "9042"))
 
-    logger.info(f"Connecting to Cassandra at {contact_points}:{port}")
+    logger.info(f"\n📡 Connecting to Cassandra at {contact_points}:{port}")
 
     # Use cluster in context manager for automatic cleanup
     async with AsyncCluster(contact_points, port=port) as cluster:
@@ -200,12 +222,15 @@ async def main():
         await demonstrate_context_manager_isolation(cluster)
         await demonstrate_concurrent_safety(cluster)
 
-    logger.info("\nAll demonstrations completed successfully!")
-    logger.info("Key takeaways:")
-    logger.info("1. Query errors don't close sessions")
-    logger.info("2. Streaming errors don't close sessions")
-    logger.info("3. Context managers only close their own resources")
-    logger.info("4. Multiple operations can safely share sessions and clusters")
+    logger.info("\n" + "=" * 80)
+    logger.info("✅ ALL DEMONSTRATIONS COMPLETED SUCCESSFULLY!")
+    logger.info("=" * 80)
+    logger.info("\n🎯 Key Takeaways:")
+    logger.info("   1. Query errors don't close sessions")
+    logger.info("   2. Streaming errors don't close sessions")
+    logger.info("   3. Context managers only close their own resources")
+    logger.info("   4. Multiple operations can safely share sessions and clusters")
+    logger.info("\n💡 Best Practice: Always use context managers for proper resource management!")
 
 
 if __name__ == "__main__":