Spaces:
Running
Running
File size: 2,799 Bytes
20cccb6 a221e9d 40403f3 20cccb6 40403f3 20cccb6 a873147 40403f3 a873147 20cccb6 5ed9749 97b3bfd 40403f3 20cccb6 40403f3 a873147 40403f3 a873147 40403f3 a873147 40403f3 a873147 |
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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
"""
test_db.py
This script verifies the database connection for the Expressive TTS Arena project.
It attempts to connect to the PostgreSQL database using async SQLAlchemy and executes
a simple query to confirm connectivity.
Functionality:
- Loads the database connection from `database.py`.
- Attempts to establish an async connection to the database.
- Executes a test query (`SELECT 1`) to confirm connectivity.
- Prints a success message if the connection is valid.
- Prints an error message if the connection fails.
Usage:
python src/test_db.py
Expected Output:
Database connection successful! (if the database is reachable)
Database connection failed: <error message> (if there are connection issues)
Troubleshooting:
- Ensure the `.env` file contains a valid `DATABASE_URL`.
- Check that the database server is running and accessible.
- Verify PostgreSQL credentials and network settings.
"""
# Standard Library Imports
import asyncio
import sys
# Third-Party Library Imports
from sqlalchemy import text
# Local Application Imports
from src.common import Config, logger
from src.database import engine, init_db
async def test_connection_async():
"""
Asynchronously test the database connection.
This function attempts to connect to the database using the configured
async engine and execute a simple SELECT query. It logs success or failure
messages accordingly.
Returns:
bool: True if the connection was successful, False otherwise.
"""
if engine is None:
logger.error("No valid database engine configured.")
return False
try:
# Create a new async session
async with engine.connect() as conn:
# Execute a simple query to verify connectivity
result = await conn.execute(text("SELECT 1"))
# Fetch the result to make sure the query completes
await result.fetchone()
logger.info("Async database connection successful!")
return True
except Exception as e:
logger.error(f"Async database connection failed: {e}")
return False
def main():
"""
Main entry point for the database connection test script.
This function creates the configuration, initializes the database engine,
and runs the async test function within an event loop. It exits with an
appropriate system exit code based on the test result.
Returns:
None
"""
# Make sure config is loaded first to initialize the engine
config = Config.get()
# Initialize the database engine
init_db(config)
# Run the async test function
success = asyncio.run(test_connection_async())
# Exit with an appropriate status code
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
|