File size: 1,686 Bytes
1d64b43 130ce2e 1d64b43 130ce2e 1d64b43 130ce2e 1d64b43 130ce2e 1d64b43 130ce2e 1d64b43 |
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 |
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)
|