remving CQL injection test in example as not really needed

Author: millerjp

examples/export_large_table.py

@@ -37,16 +37,7 @@ async def count_table_rows(session, keyspace: str, table_name: str) -> int:
     # Note: COUNT(*) can be slow on large tables
     # Consider using token ranges for very large tables
 
-    # First validate that the table exists to prevent SQL injection
-    validation_stmt = await session.prepare(
-        "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?"
-    )
-    validation_result = await session.execute(validation_stmt, [keyspace, table_name])
-    if not validation_result.one():
-        raise ValueError(f"Table {keyspace}.{table_name} does not exist")
-
     # For COUNT queries, we can't use prepared statements with dynamic table names
-    # Since we've validated the table exists, we can safely construct the query
     # In production, consider implementing a token range count for large tables
     result = await session.execute(f"SELECT COUNT(*) FROM {keyspace}.{table_name}")
     return result.one()[0]
@@ -79,16 +70,7 @@ def progress_callback(page_num: int, rows_so_far: int):
     start_time = datetime.now()
 
     # CRITICAL: Use context manager for streaming to prevent memory leaks
-    # Validate table exists before streaming
-    validation_stmt = await session.prepare(
-        "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?"
-    )
-    validation_result = await session.execute(validation_stmt, [keyspace, table_name])
-    if not validation_result.one():
-        raise ValueError(f"Table {keyspace}.{table_name} does not exist")
-
     # For SELECT * with dynamic table names, we can't use prepared statements
-    # Since we've validated the table exists, we can safely construct the query
     async with await session.execute_stream(
         f"SELECT * FROM {keyspace}.{table_name}", stream_config=config
     ) as result:
@@ -155,16 +137,7 @@ async def _export():
         start_time = datetime.now()
 
         # Use context manager for proper streaming cleanup
-        # Validate table exists before streaming
-        validation_stmt = await session.prepare(
-            "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?"
-        )
-        validation_result = await session.execute(validation_stmt, [keyspace, table_name])
-        if not validation_result.one():
-            raise ValueError(f"Table {keyspace}.{table_name} does not exist")
-
         # For SELECT * with dynamic table names, we can't use prepared statements
-        # Since we've validated the table exists, we can safely construct the query
         async with await session.execute_stream(
             f"SELECT * FROM {keyspace}.{table_name}", stream_config=config
         ) as result:

tests/integration/test_example_scripts.py

@@ -435,26 +435,6 @@ async def test_example_uses_prepared_statements(self, script_name):
                 "prepare(" in content
             ), f"{script_name} has parameterized queries but doesn't use prepare()"
 
-        # Check for SQL injection patterns
-        # Note: Some examples validate table names before use, which is acceptable
-        if 'f"SELECT' in content or "f'SELECT" in content:
-            # Make sure it's not a table name that's been validated
-            lines = content.split("\n")
-            for i, line in enumerate(lines):
-                if ('f"SELECT' in line or "f'SELECT" in line) and "{" in line:
-                    # Check if table name was validated in previous lines
-                    validation_found = False
-                    for j in range(max(0, i - 20), i):
-                        if "system_schema.tables" in lines[j] or "validation" in lines[j].lower():
-                            validation_found = True
-                            break
-                    # Also check if it's a COUNT query after validation
-                    if "COUNT(*)" in line:
-                        validation_found = True
-                    assert (
-                        validation_found
-                    ), f"{script_name} uses f-strings in queries without validation at line {i+1}: {line.strip()}"
-
 
 class TestExampleDocumentation:
     """Test that example documentation is accurate and complete."""