| 12345678910111213141516171819202122232425262728293031323334353637383940 |
- import os
- from dotenv import load_dotenv
- import psycopg2
- load_dotenv()
- def test_tool_provider():
- print("Testing connection to tool_provider table...")
- try:
- conn = psycopg2.connect(
- host=os.getenv('KNOWHUB_DB'),
- port=int(os.getenv('KNOWHUB_PORT', 5432)),
- user=os.getenv('KNOWHUB_USER'),
- password=os.getenv('KNOWHUB_PASSWORD'),
- database=os.getenv('KNOWHUB_DB_NAME')
- )
- c = conn.cursor()
-
- # Check if tool_provider table exists by selecting 0 rows
- c.execute("SELECT * FROM tool_provider LIMIT 0")
- print("Success! Table tool_provider exists.")
-
- # Ensure it has the correct columns
- expected_cols = ['tool_id', 'provider_id']
- col_names = [desc[0] for desc in c.description]
- print(f"Columns in tool_provider: {col_names}")
-
- # Check if there's any data
- c.execute("SELECT * FROM tool_provider LIMIT 5")
- rows = c.fetchall()
- print(f"Sample data from tool_provider: {rows}")
-
- cursor = conn.cursor()
- c.close()
- conn.close()
- except Exception as e:
- print(f"Failed to access tool_provider table: {e}")
- if __name__ == "__main__":
- test_tool_provider()
|