Spaces:
Sleeping
Sleeping
File size: 2,566 Bytes
6bd3e57 f4f2765 6bd3e57 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | from sales_assistant.db.utils.db_connections import validate_config
from sales_assistant.db.utils.db_utils import read_sql, write_to_table
import pandas as pd
# Validate configuration
required_vars = [
'SSH_HOSTNAME', 'SSH_USERNAME', 'SSH_PEM_CONTENT',
'MYSQL_HOST', 'MYSQL_USER', 'MYSQL_PASSWORD', 'MYSQL_DB'
]
validate_config_vars = validate_config(required_vars)
print("Config validation:", validate_config_vars)
# Test basic connection
print("\n=== Testing basic connection ===")
test_connection = read_sql("SELECT 1 as test_column;")
print("Connection test result:")
print(test_connection)
# Test reading from products table
print("\n=== Testing products table read ===")
test_product_table = read_sql("SELECT * FROM streamnet.products_list LIMIT 5;")
print("Products table sample:")
print(test_product_table.head())
print(f"Shape: {test_product_table.shape}")
# Test writing a sample DataFrame
print("\n=== Testing DataFrame write (create new table) ===")
sample_data = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Test Product 1', 'Test Product 2', 'Test Product 3'],
'price': [10.99, 20.50, 15.75],
'description': ['Sample description 1', 'Sample description 2', 'Sample description 3']
})
print("Sample data to write:")
print(sample_data)
# Create new table using convenience function
success = write_to_table(
df=sample_data,
table_name='test_products',
schema='streamnet',
if_exists='replace' # Use replace to create new table
)
if success:
print("✅ Create table test successful!")
# Test appending more data
print("\n=== Testing DataFrame append ===")
additional_data = pd.DataFrame({
'id': [4, 5],
'name': ['Test Product 4', 'Test Product 5'],
'price': [25.00, 30.99],
'description': ['Sample description 4', 'Sample description 5']
})
print("Additional data to append:")
print(additional_data)
append_success = write_to_table(
df=additional_data,
table_name='test_products',
schema='streamnet',
if_exists='append'
)
if append_success:
print("✅ Append test successful!")
# Read back all the data
print("\n=== Verifying all written data ===")
read_back = read_sql("SELECT * FROM streamnet.test_products ORDER BY id;")
print("All data in test table:")
print(read_back)
print(f"Total rows: {len(read_back)}")
else:
print("❌ Append test failed!")
else:
print("❌ Create table test failed!") |