Create test_connection.py

mongodb/test_connection.py

@@ -0,0 +1,85 @@
+"""
+Test MongoDB Connection Script
+Run this first to verify your MongoDB connection is working before inserting sample data.
+
+Usage: python test_connection.py
+"""
+
+import os
+from pymongo import MongoClient
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+def test_mongodb_connection():
+    """Test MongoDB connection using environment variables"""
+    try:
+        # Get MongoDB URL from environment variables (same as your utils.py)
+        mongodb_url = os.getenv('MONGODB_URL') or os.getenv('MONGO_URL') or os.getenv('DATABASE_URL')
+        
+        if not mongodb_url:
+            print("❌ Error: No MongoDB URL found in environment variables")
+            print("Please add one of these to your .env file:")
+            print("  MONGODB_URL=your_mongodb_connection_string")
+            print("  MONGO_URL=your_mongodb_connection_string") 
+            print("  DATABASE_URL=your_mongodb_connection_string")
+            return False
+        
+        print(f"🔄 Connecting to MongoDB...")
+        print(f"URL: {mongodb_url[:20]}{'...' if len(mongodb_url) > 20 else ''}")
+        
+        # Connect to MongoDB
+        client = MongoClient(mongodb_url, serverSelectionTimeoutMS=5000)
+        
+        # Test the connection
+        client.admin.command('ping')
+        print("✅ Successfully connected to MongoDB!")
+        
+        # Get database name
+        db_name = os.getenv('MONGODB_DATABASE', 'sparrow_logistics')
+        db = client[db_name]
+        print(f"✅ Database: {db_name}")
+        
+        # List existing collections
+        collections = db.list_collection_names()
+        if collections:
+            print(f"📁 Existing collections: {', '.join(collections)}")
+        else:
+            print("📁 No existing collections found (this is normal for a new database)")
+        
+        # Test write permission
+        test_collection = db.connection_test
+        test_doc = {"test": "connection", "timestamp": "now"}
+        result = test_collection.insert_one(test_doc)
+        print(f"✅ Write test successful (inserted document: {result.inserted_id})")
+        
+        # Clean up test document
+        test_collection.delete_one({"_id": result.inserted_id})
+        print("✅ Cleanup successful")
+        
+        client.close()
+        print("🎉 MongoDB connection test completed successfully!")
+        print("You can now run the sample data insertion script.")
+        return True
+        
+    except Exception as e:
+        print(f"❌ MongoDB connection failed: {e}")
+        print("\nTroubleshooting tips:")
+        print("1. Check your .env file has the correct MONGODB_URL")
+        print("2. Verify your MongoDB server is running")
+        print("3. Check your network connection")
+        print("4. Verify your MongoDB credentials are correct")
+        return False
+
+if __name__ == "__main__":
+    print("MongoDB Connection Test for Sparrow Logistics")
+    print("=" * 50)
+    
+    success = test_mongodb_connection()
+    
+    if success:
+        print("\n✅ Ready to proceed with sample data insertion!")
+    else:
+        print("\n❌ Please fix the connection issues before proceeding.")
+        exit(1)