from flask import Flask, jsonify import psycopg2 import os import socket app = Flask(__name__) def check_port(host: str, port: int): """Check if a specific port on a host is open.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.settimeout(1) # 1 second timeout try: sock.connect((host, port)) return True except socket.error: return False def connect_db(): """Attempt to connect to the PostgreSQL database and retrieve version info.""" try: connection = psycopg2.connect( dbname=os.getenv('POSTGRES_DB'), user=os.getenv('POSTGRES_USER'), password=os.getenv('POSTGRES_PASSWORD'), host='localhost' ) cursor = connection.cursor() cursor.execute("SELECT version();") db_version = cursor.fetchone() return { "status": "success", "db_version": db_version, "user": os.getenv('POSTGRES_USER'), "database": os.getenv('POSTGRES_DB') } except Exception as e: return {"status": "failure", "error": str(e)} @app.route('/') def home(): # Check if PostgreSQL is running and port is open db_running = check_port('localhost', 5432) # Perform database connection test db_info = connect_db() result = { "message": "Database Connection Test Results", "database_running": db_running, "port_open": db_running, # The same check, for clarity "connection_result": db_info } return jsonify(result) if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)