|
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) |
|
|