| 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) |
| 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(): |
| |
| db_running = check_port('localhost', 5432) |
| |
| |
| db_info = connect_db() |
| |
| result = { |
| "message": "Database Connection Test Results", |
| "database_running": db_running, |
| "port_open": db_running, |
| "connection_result": db_info |
| } |
| |
| return jsonify(result) |
|
|
| if __name__ == '__main__': |
| app.run(host='0.0.0.0', port=7860) |
|
|