updating examples and docs

Author: millerjp

README.md

@@ -50,23 +50,33 @@ async def get_user(user_id: str):
 
 ```python
 # ✅ Async code with our wrapper - NON-BLOCKING
+from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from async_cassandra import AsyncCluster
 
-app = FastAPI()
-cluster = AsyncCluster(['localhost'])
 session = None
-
-@app.on_event("startup")
-async def startup():
-    global session
-    session = await cluster.connect()
+user_stmt = None
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global session, user_stmt
+    # Use context managers for proper resource management
+    async with AsyncCluster(['localhost']) as cluster:
+        async with cluster.connect() as session:
+            # Prepare statement for better performance
+            user_stmt = await session.prepare(
+                "SELECT * FROM users WHERE id = ?"
+            )
+            yield  # App runs here
+            # Cleanup happens automatically
+
+app = FastAPI(lifespan=lifespan)
 
 @app.get("/users/{user_id}")
 async def get_user(user_id: str):
     # This doesn't block! The event loop remains free to handle
     # other requests while waiting for the database response
-    result = await session.execute("SELECT * FROM users WHERE id = %s", [user_id])
+    result = await session.execute(user_stmt, [user_id])
     return {"user": result.one()}
 ```
 

docs/getting-started.md

@@ -27,21 +27,17 @@ import asyncio
 from async_cassandra import AsyncCluster
 
 async def main():
-    # 1. Create a cluster object (doesn't connect yet)
-    cluster = AsyncCluster(['localhost'])
-
-    # 2. Connect and get a session (this is where connection happens)
-    session = await cluster.connect('my_keyspace')
-
-    # 3. Execute a query (waits for results without blocking)
-    result = await session.execute("SELECT * FROM users LIMIT 10")
+    # Use context managers for automatic cleanup (REQUIRED)
+    async with AsyncCluster(['localhost']) as cluster:
+        async with cluster.connect('my_keyspace') as session:
+            # Execute a query (waits for results without blocking)
+            result = await session.execute("SELECT * FROM users LIMIT 10")
 
-    # 4. Process results (rows are like dictionaries)
-    for row in result:
-        print(f"User: {row.name}, Email: {row.email}")
+            # Process results (rows are like dictionaries)
+            for row in result:
+                print(f"User: {row.name}, Email: {row.email}")
 
-    # 5. Clean up (IMPORTANT: always close connections)
-    await cluster.shutdown()
+            # No manual cleanup needed - context managers handle it!
 
 # Run the async function
 asyncio.run(main())
@@ -353,42 +349,39 @@ FastAPI is an async web framework. If you use the regular Cassandra driver, it w
 
 ### FastAPI Example
 
-#### The Setup
+#### The Setup (Using Lifespan Context Manager)
 
 ```python
+from contextlib import asynccontextmanager
 from fastapi import FastAPI, HTTPException
 from async_cassandra import AsyncCluster
 import uuid
 
-app = FastAPI()
-
-# Global variables for cluster and session
-cluster = None
+# Global variables for session and prepared statements
 session = None
 prepared_statements = {}
 
-@app.on_event("startup")
-async def startup():
-    """Initialize Cassandra connection when server starts."""
-    global cluster, session, prepared_statements
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Manage Cassandra connection lifecycle."""
+    global session, prepared_statements
 
-    # Create cluster connection
-    cluster = AsyncCluster(['localhost'])
-    session = await cluster.connect('my_keyspace')
+    # Startup: Create connection with context manager
+    async with AsyncCluster(['localhost']) as cluster:
+        async with cluster.connect('my_keyspace') as session:
+            # Prepare statements once at startup (more efficient)
+            prepared_statements['get_user'] = await session.prepare(
+                "SELECT * FROM users WHERE id = ?"
+            )
+            prepared_statements['create_user'] = await session.prepare(
+                "INSERT INTO users (id, name, email) VALUES (?, ?, ?)"
+            )
 
-    # Prepare statements once at startup (more efficient)
-    prepared_statements['get_user'] = await session.prepare(
-        "SELECT * FROM users WHERE id = ?"
-    )
-    prepared_statements['create_user'] = await session.prepare(
-        "INSERT INTO users (id, name, email) VALUES (?, ?, ?)"
-    )
+            yield  # Server runs here
+
+            # Cleanup happens automatically when context exits
 
-@app.on_event("shutdown")
-async def shutdown():
-    """Clean up when server stops."""
-    if cluster:
-        await cluster.shutdown()
+app = FastAPI(lifespan=lifespan)
 ```
 
 #### API Endpoints

examples/README.md

@@ -82,9 +82,46 @@ python metrics_simple.py
 
 ### 6. [Advanced Metrics](metrics_example.py)
 
-More comprehensive metrics example (requires updates to work with current API).
+Comprehensive metrics and observability example:
+- Multiple metrics collectors setup
+- Query performance monitoring
+- Connection health tracking
+- Prometheus integration example
+- FastAPI integration patterns
 
-### 7. [Monitoring Configuration](monitoring/)
+**Run:**
+```bash
+python metrics_example.py
+```
+
+### 7. [Context Manager Safety](context_manager_safety_demo.py)
+
+Demonstrates proper context manager usage:
+- Context manager isolation
+- Error safety in queries and streaming
+- Concurrent operations with shared resources
+- Resource cleanup guarantees
+
+**Run:**
+```bash
+python context_manager_safety_demo.py
+```
+
+### 8. [Thread Pool Configuration](thread_pool_configuration.py)
+
+Shows how to configure thread pools for different workloads:
+- Web application configuration
+- Batch processing setup
+- Thread pool size comparison
+- Thread starvation demonstration
+- Thread pool monitoring
+
+**Run:**
+```bash
+python thread_pool_configuration.py
+```
+
+### 9. [Monitoring Configuration](monitoring/)
 
 Production-ready monitoring configurations:
 - **alerts.yml** - Prometheus alerting rules for:
@@ -114,25 +151,48 @@ All examples require:
    # pip install async-cassandra
    ```
 
-## Common Patterns Demonstrated
+## Best Practices Demonstrated
+
+### MANDATORY: Always Use Context Managers
+All examples follow the required pattern:
+```python
+# ALWAYS use context managers for resource management
+async with AsyncCluster(["localhost"]) as cluster:
+    async with cluster.connect() as session:
+        # Your code here
+        pass
+```
+
+### MANDATORY: Always Use PreparedStatements
+For any query with parameters:
+```python
+# Prepare statement once
+stmt = await session.prepare(
+    "INSERT INTO users (id, name) VALUES (?, ?)"
+)
+# Execute many times
+await session.execute(stmt, [user_id, name])
+```
+
+### Common Patterns Demonstrated
 
-### Connection Management
-- Using context managers for automatic cleanup
+#### Connection Management
+- Using context managers for automatic cleanup (REQUIRED)
 - Proper cluster and session lifecycle
 - Connection health monitoring
 
-### Error Handling
+#### Error Handling
 - Catching and handling Cassandra exceptions
 - Retry strategies with idempotency
 - Graceful degradation
 
-### Performance Optimization
-- Prepared statements for repeated queries
+#### Performance Optimization
+- Prepared statements for repeated queries (REQUIRED)
 - Concurrent query execution
-- Streaming for large datasets
+- Streaming for large datasets with context managers
 - Appropriate fetch sizes
 
-### Monitoring & Observability
+#### Monitoring & Observability
 - Metrics collection
 - Performance tracking
 - Health checks
@@ -159,11 +219,14 @@ Examples use local Cassandra by default. Network latency may vary with remote cl
 ## Contributing
 
 We welcome new examples! When contributing:
+- **MUST use context managers** for all cluster/session/streaming operations
+- **MUST use PreparedStatements** for all parameterized queries
 - Include clear documentation in the code
 - Handle errors appropriately
 - Clean up resources (drop keyspaces/tables)
 - Test with Python 3.12
 - Update this README
+- Follow the patterns shown in existing examples
 
 ## Support
 

examples/context_manager_safety_demo.py

@@ -65,11 +65,10 @@ async def demonstrate_streaming_error_safety(cluster):
         """
     )
 
-    # Insert some data
+    # Insert some data using prepared statement
+    insert_stmt = await session.prepare("INSERT INTO test_data (id, value) VALUES (?, ?)")
     for i in range(10):
-        await session.execute(
-            "INSERT INTO test_data (id, value) VALUES (%s, %s)", [uuid.uuid4(), f"value_{i}"]
-        )
+        await session.execute(insert_stmt, [uuid.uuid4(), f"value_{i}"])
 
     # Try streaming from non-existent table (will fail)
     try:

examples/export_large_table.py

@@ -283,35 +283,29 @@ async def setup_sample_data(session):
 
 async def main():
     """Run the export example."""
-    # Connect to Cassandra
-    cluster = AsyncCluster(["localhost"])
-
-    try:
-        session = await cluster.connect()
-
-        # Setup sample data
-        await setup_sample_data(session)
-
-        # Create output directory
-        output_dir = Path("exports")
-        output_dir.mkdir(exist_ok=True)
-
-        # Export using async I/O if available
-        if ASYNC_FILE_IO:
-            await export_table_async(
-                session, "export_example", "products", str(output_dir / "products_async.csv")
-            )
-        else:
-            await export_table_sync(
-                session, "export_example", "products", str(output_dir / "products_sync.csv")
-            )
-
-        # Cleanup (optional)
-        logger.info("\nCleaning up...")
-        await session.execute("DROP KEYSPACE export_example")
+    # Connect to Cassandra using context manager
+    async with AsyncCluster(["localhost"]) as cluster:
+        async with cluster.connect() as session:
+            # Setup sample data
+            await setup_sample_data(session)
+
+            # Create output directory
+            output_dir = Path("exports")
+            output_dir.mkdir(exist_ok=True)
+
+            # Export using async I/O if available
+            if ASYNC_FILE_IO:
+                await export_table_async(
+                    session, "export_example", "products", str(output_dir / "products_async.csv")
+                )
+            else:
+                await export_table_sync(
+                    session, "export_example", "products", str(output_dir / "products_sync.csv")
+                )
 
-    finally:
-        await cluster.shutdown()
+            # Cleanup (optional)
+            logger.info("\nCleaning up...")
+            await session.execute("DROP KEYSPACE export_example")
 
 
 if __name__ == "__main__":

examples/metrics_example.py

@@ -28,13 +28,10 @@ async def main():
     )
 
     # 2. Create cluster and session with metrics
-    cluster = AsyncCluster(contact_points=["localhost"])
-    session = await cluster.connect()
-
-    # Inject metrics middleware into session
-    session._metrics = metrics
-
-    try:
+    async with AsyncCluster(contact_points=["localhost"]) as cluster:
+        async with cluster.connect() as session:
+            # Inject metrics middleware into session
+            session._metrics = metrics
         # 3. Set up test environment
         await session.execute(
             """
@@ -135,10 +132,6 @@ async def main():
                     print(f"  Total Queries: {health['total_queries']}")
                     print(f"  Errors: {health['error_count']}")
 
-    finally:
-        await session.close()
-        await cluster.shutdown()
-
 
 async def prometheus_example():
     """Example showing Prometheus integration."""
@@ -153,20 +146,20 @@ async def prometheus_example():
         print("📊 Prometheus metrics server started on http://localhost:8000")
 
         # Run some queries with metrics
-        cluster = AsyncCluster(contact_points=["localhost"])
-        session = await cluster.connect()
-        session._metrics = metrics
+        async with AsyncCluster(contact_points=["localhost"]) as cluster:
+            async with cluster.connect() as session:
+                session._metrics = metrics
 
-        # Execute some queries
-        for i in range(5):
-            await session.execute("SELECT release_version FROM system.local")
+                # Prepare statement
+                stmt = await session.prepare("SELECT release_version FROM system.local")
 
-        print("📈 Metrics available at http://localhost:8000/metrics")
-        print("Sample metrics output:")
-        print(generate_latest().decode("utf-8")[:500] + "...")
+                # Execute some queries
+                for i in range(5):
+                    await session.execute(stmt, [])
 
-        await session.close()
-        await cluster.shutdown()
+                print("📈 Metrics available at http://localhost:8000/metrics")
+                print("Sample metrics output:")
+                print(generate_latest().decode("utf-8")[:500] + "...")
 
     except ImportError:
         print("❌ prometheus_client not installed. Install with: pip install prometheus_client")