Test both database connections
(logger)
| 30 | |
| 31 | |
| 32 | def test_connections(logger): |
| 33 | """Test both database connections""" |
| 34 | logger.info("Testing database connections...") |
| 35 | |
| 36 | # Test PostgreSQL connection |
| 37 | try: |
| 38 | from sqlalchemy import text |
| 39 | |
| 40 | from questions.db_models_postgres import SessionLocal, create_tables |
| 41 | |
| 42 | db = SessionLocal() |
| 43 | db.execute(text("SELECT 1")) |
| 44 | db.close() |
| 45 | logger.info("✅ PostgreSQL connection successful") |
| 46 | |
| 47 | # Ensure tables exist |
| 48 | create_tables() |
| 49 | logger.info("✅ PostgreSQL tables created/verified") |
| 50 | |
| 51 | except Exception as e: |
| 52 | logger.error(f"❌ PostgreSQL connection failed: {e}") |
| 53 | return False |
| 54 | |
| 55 | # Test Google Cloud NDB connection |
| 56 | try: |
| 57 | from google.cloud import ndb |
| 58 | |
| 59 | # Define NDB User model for migration |
| 60 | class NDBUser(ndb.Model): |
| 61 | id = ndb.StringProperty(required=True) |
| 62 | cookie_user = ndb.IntegerProperty() |
| 63 | created = ndb.DateTimeProperty(auto_now_add=True) |
| 64 | updated = ndb.DateTimeProperty(auto_now=True) |
| 65 | is_subscribed = ndb.BooleanProperty(default=False) |
| 66 | num_self_hosted_instances = ndb.IntegerProperty(default=0) |
| 67 | name = ndb.StringProperty() |
| 68 | email = ndb.StringProperty() |
| 69 | profile_url = ndb.StringProperty() |
| 70 | access_token = ndb.StringProperty() |
| 71 | photo_url = ndb.StringProperty() |
| 72 | stripe_id = ndb.StringProperty() |
| 73 | secret = ndb.StringProperty() |
| 74 | free_credits = ndb.IntegerProperty(default=0) |
| 75 | charges_monthly = ndb.IntegerProperty(default=0) |
| 76 | |
| 77 | client = ndb.Client(project=os.environ.get("GOOGLE_CLOUD_PROJECT", "questions-346919")) |
| 78 | |
| 79 | with client.context(): |
| 80 | # Try to query one user to test connection |
| 81 | NDBUser.query().fetch(1) |
| 82 | logger.info("✅ Google Cloud NDB connection successful") |
| 83 | |
| 84 | except Exception as e: |
| 85 | logger.error(f"❌ Google Cloud NDB connection failed: {e}") |
| 86 | logger.error("Make sure GOOGLE_APPLICATION_CREDENTIALS is set correctly") |
| 87 | return False |
| 88 | |
| 89 | return True |
no test coverage detected